首页 > 软件编程 > java > RabbitMQ多路复用Channel

RabbitMQ消息队列中多路复用Channel信道详解

2023-08-28 11:44:23 作者：swordbob

这篇文章主要介绍了RabbitMQ消息队列中多路复用Channel信道详解,消息Message是指在应用间传送的数据,消息可以非常简单,比如只包含文本字符串,也可以更复杂,可能包含嵌入对象,需要的朋友可以参考下

什么叫消息队列

消息（Message）是指在应用间传送的数据。

消息可以非常简单，比如只包含文本字符串，也可以更复杂，可能包含嵌入对象。

消息队列（Message Queue）是一种应用间的通信方式，消息发送后可以立即返回，由消息系统来确保消息的可靠传递。

消息发布者只管把消息发布到 MQ 中而不用管谁来取，消息使用者只管从 MQ 中取消息而不管是谁发布的。

这样发布者和使用者都不用知道对方的存在。

最简单的理解是开了两个定时器的线程，分别是上传参数和下载数据两个不同的线程，他们之间通过数据库（也就是这里的queue）进行异步联通，但这里并不是用的定时器而是用的观察者模式

RabbitMQ 基本概念

上面只是最简单抽象的描述，具体到 RabbitMQ 则有更详细的概念需要解释。

上面介绍过 RabbitMQ 是 AMQP 协议的一个开源实现，所以其内部实际上也是 AMQP 中的基本概念：

RabbitMQ 内部结构

Message
- 消息，消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可能需要持久性存储）等。
Publisher
- 消息的生产者，也是一个向交换器发布消息的客户端应用程序。
Exchange
- 交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。
Binding
- 绑定，用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。
Queue
- 消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。
Connection
- 网络连接，比如一个TCP连接。
Channel
- 信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内地虚拟连接，AMQP 命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。
Consumer
- 消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。
Virtual Host
- 虚拟主机，表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥有自己的队列、交换器、绑定和权限机制。vhost 是 AMQP 概念的基础，必须在连接时指定，RabbitMQ 默认的 vhost 是 / 。
Broker
- 表示消息队列服务器实体。

对于理解多路复用，需要讲下NIO:

其实，多路复用是一种思想，多路是指多个客户端连接线路（TCP、Channel），复用是指使用一个线程重复使用，总的来说，就是单线程能同时处理多个请求。要实现这一点，就得改造BIO的连接模式了，BIO是客户端直接连接服务端，NIO采用的是多路复用器 (Selector)，相当于客户端的连接不会直接连接服务端，而是连接到多路复用器。

这样做的好处，就是把服务端和客户端隔离了，如果不直接连接，服务器端就不会阻塞，多路复用器会将收到的消息做为事件请求发送给服务端，但服务端在处理事件的时候对于其他客户端来说还是阻塞的，这些事件有不同类型。

Channel

了解了多路复用这个设计后，再讲下Channel部分，Channel是一个双向读写通道，是异步传输的，基于数据块结构传输，BIO使用的是Stream流，基于字节传输。

关于性能方面，我也没做过测试，但我知道一口咬定NIO比BIO性能要高效的言论，是错误的，NIO主要解决的不是性能的问题 Channel有很多实现，因为本文介绍的是Socket，客户端使用SocketChannel，服务端使用ServerSocketChannel

代码实现

服务端：

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
public class NioServer {
    public static void main(String[] args) throws IOException, InterruptedException {
        // 打开Channel服务端并绑定监听一个端口
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.socket().bind(new InetSocketAddress(8459));
        ssc.configureBlocking(false);
        // 打开多路复用器 并注册到 ServerSocketChannel 并监听连接事件
        Selector selector = Selector.open();
        ssc.register(selector, SelectionKey.OP_ACCEPT);
        System.out.println("服务器已启动...");
        while (true) {
            // 如果没有事件发生 则select() 处于阻塞状态
            selector.select();
            // 发生事件
            Set<SelectionKey> selectionKeys = selector.selectedKeys();
            Iterator<SelectionKey> iterator = selectionKeys.iterator();
            // 处理事件
            while (iterator.hasNext()) {
                // 拿到具体事件
                SelectionKey selectionKey = iterator.next();
                // 判断事件的类型
                if (selectionKey.isAcceptable()) {
                    System.out.println("客户端连接事件");
                    SocketChannel channel = ssc.accept();
                    channel.configureBlocking(false);
                    channel.register(selector, SelectionKey.OP_READ);
                    if (channel.finishConnect()) {
                        System.out.println("完成连接");
                    }
                } else if (selectionKey.isReadable()) {
                    SocketChannel sc = (SocketChannel) selectionKey.channel();
                    ByteBuffer buffer = ByteBuffer.allocate(1024);
                    int read = sc.read(buffer);
                    System.out.println("收到的消息:" + new String(buffer.array(), 0, read));
                    // 响应客户端  这里可有可无
                    buffer.clear();
                    buffer.put("已收到消息".getBytes());
                    // 将缓冲区各标志复位，因为向里面put了数据标志被改变要想从中读取数据发向服务器，就要复位
                    buffer.flip();
                    sc.write(buffer);
                    // 设置监听事件的集合 这里把写入事件加入
                    selectionKey.interestOps(selectionKey.interestOps() | SelectionKey.OP_WRITE);
                    System.out.println("服务器向客户端发送已确认消息");
                } else if (selectionKey.isWritable()) {
                    System.out.println("触发往客户端写入数据事件");
                    // 发送完了就取消监听写事件，否则会无限循环触发写事件
                    selectionKey.interestOps(selectionKey.interestOps() & ~SelectionKey.OP_WRITE);
                }
                // 事件完成后，将其移除
                iterator.remove();
            }
        }
    }

客户端

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Scanner;
import java.util.Set;
public class NioClient {
    public static void main(String[] args) throws IOException {
        //打开选择器
        Selector selector = Selector.open();
        //打开通道
        SocketChannel socketChannel = SocketChannel.open();
        //配置非阻塞模型
        socketChannel.configureBlocking(false);
        //连接远程主机
        socketChannel.connect(new InetSocketAddress("127.0.0.1", 8459));
        //注册事件
        socketChannel.register(selector, SelectionKey.OP_CONNECT | SelectionKey.OP_READ );
        //循环处理
        new Thread(() -> {
            while (true) {
                try {
                    selector.select();
                    Set<SelectionKey> keys = selector.selectedKeys();
                    Iterator<SelectionKey> iter = keys.iterator();
                    while (iter.hasNext()) {
                        SelectionKey key = iter.next();
                        if (key.isConnectable()) {
                            //连接建立或者连接建立不成功
                            SocketChannel channel = (SocketChannel) key.channel();
                            //完成连接的建立
                            if (channel.finishConnect()) {
                                System.out.println("完成连接");
                            }
                        } else if (key.isReadable()) {
                            SocketChannel channel = (SocketChannel) key.channel();
                            ByteBuffer buffer = ByteBuffer.allocate(1024);
                            buffer.clear();
                            channel.read(buffer);
                            System.out.println("客户端收到消息:" + new String(buffer.array()));
                            key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
                        } else if (key.isWritable()) {
                            System.out.println("客户端向服务端写入数据");
                            // 设置监听事件的集合 这里把写入事件加入
                            key.interestOps(key.interestOps() & ~SelectionKey.OP_WRITE);
                        }
                        iter.remove();
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                    break;
                }
            }
        }).start();
        Scanner scanner = new Scanner(System.in);
        while (true) {
            System.out.println("请输入要发送的字符串");
            String str = scanner.nextLine();
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            buffer.put(str.getBytes());
            buffer.flip();
            socketChannel.write(buffer);
        }
    }
}

讲完多路复用，接下来就是TCP实现的多路复用，整体和NIO很相似，都是把请求注册到selector中，根据SelectionKey进行阻塞式的通信，因为有了注册和SelectionKey使得单线程能同时处理多个请求成为可能

IO多路复用(multiplexing)属于同步IO网络模型

是以Reactor模式实现

常见的IO多路复用应用有：select、poll、epoll

本篇文章采用Java的NIO框架来实现单线程的IO多路复用

Reactor模式的组成角色

1. Reactor：负责派发IO事件给对应的角色处理。为了监听IO事件，select必须实现在Reactor中。

2. Acceptor：负责接受client的连线，然后给client绑定一个Handler并注册IO事件到Reactor上监听。

3. Handler：负责处理与client交互的事件或行为。通常因为Handler要处理与所对应client交互的多个事件或行为，为了简化设计，会以状态模式来实现Handler。

代码实现

[TCPReactor.java]

// Reactor線程  
package server;  
import java.io.IOException;  
import java.net.InetSocketAddress;  
import java.nio.channels.SelectionKey;  
import java.nio.channels.Selector;  
import java.nio.channels.ServerSocketChannel;  
import java.util.Iterator;  
import java.util.Set;  
public class TCPReactor implements Runnable {  
    private final ServerSocketChannel ssc;  
    private final Selector selector;  
    public TCPReactor(int port) throws IOException {  
        selector = Selector.open();  
        ssc = ServerSocketChannel.open();  
        InetSocketAddress addr = new InetSocketAddress(port);  
        ssc.socket().bind(addr); // 在ServerSocketChannel綁定監聽端口  
        ssc.configureBlocking(false); // 設置ServerSocketChannel為非阻塞  
        SelectionKey sk = ssc.register(selector, SelectionKey.OP_ACCEPT); // ServerSocketChannel向selector註冊一個OP_ACCEPT事件，然後返回該通道的key  
        sk.attach(new Acceptor(selector, ssc)); // 給定key一個附加的Acceptor對象  
    }  
    @Override  
    public void run() {  
        while (!Thread.interrupted()) { // 在線程被中斷前持續運行  
            System.out.println("Waiting for new event on port: " + ssc.socket().getLocalPort() + "...");  
            try {  
                if (selector.select() == 0) // 若沒有事件就緒則不往下執行  
                    continue;  
            } catch (IOException e) {  
                // TODO Auto-generated catch block  
                e.printStackTrace();  
            }  
            Set<SelectionKey> selectedKeys = selector.selectedKeys(); // 取得所有已就緒事件的key集合  
            Iterator<SelectionKey> it = selectedKeys.iterator();  
            while (it.hasNext()) {  
                dispatch((SelectionKey) (it.next())); // 根據事件的key進行調度  
                it.remove();  
            }  
        }  
    }  
    /* 
     * name: dispatch(SelectionKey key) 
     * description: 調度方法，根據事件綁定的對象開新線程 
     */  
    private void dispatch(SelectionKey key) {  
        Runnable r = (Runnable) (key.attachment()); // 根據事件之key綁定的對象開新線程  
        if (r != null)  
            r.run();  
    }  
}

[Acceptor.java]

// 接受連線請求線程  
package server;  
import java.io.IOException;  
import java.nio.channels.SelectionKey;  
import java.nio.channels.Selector;  
import java.nio.channels.ServerSocketChannel;  
import java.nio.channels.SocketChannel;  
public class Acceptor implements Runnable {  
    private final ServerSocketChannel ssc;  
    private final Selector selector;  
    public Acceptor(Selector selector, ServerSocketChannel ssc) {  
        this.ssc=ssc;  
        this.selector=selector;  
    }  
    @Override  
    public void run() {  
        try {  
            SocketChannel sc= ssc.accept(); // 接受client連線請求  
            System.out.println(sc.socket().getRemoteSocketAddress().toString() + " is connected.");  
            if(sc!=null) {  
                sc.configureBlocking(false); // 設置為非阻塞  
                SelectionKey sk = sc.register(selector, SelectionKey.OP_READ); // SocketChannel向selector註冊一個OP_READ事件，然後返回該通道的key  
                selector.wakeup(); // 使一個阻塞住的selector操作立即返回  
                sk.attach(new TCPHandler(sk, sc)); // 給定key一個附加的TCPHandler對象  
            }  
        } catch (IOException e) {  
            // TODO Auto-generated catch block  
            e.printStackTrace();  
        }  
    }  
}

我们先来简单点的，Handler不以状态模式实现，只以比较直觉的方式实现。

[TCPHandler.java]

// Handler線程  
package server;  
import java.io.IOException;  
import java.nio.ByteBuffer;  
import java.nio.channels.SelectionKey;  
import java.nio.channels.SocketChannel;  
import java.util.concurrent.LinkedBlockingQueue;  
import java.util.concurrent.ThreadPoolExecutor;  
import java.util.concurrent.TimeUnit;  
public class TCPHandler implements Runnable {  
    private final SelectionKey sk;  
    private final SocketChannel sc;  
    int state;   
    public TCPHandler(SelectionKey sk, SocketChannel sc) {  
        this.sk = sk;  
        this.sc = sc;  
        state = 0; // 初始狀態設定為READING  
    }  
    @Override  
    public void run() {  
        try {  
            if (state == 0)  
                read(); // 讀取網絡數據  
            else  
                send(); // 發送網絡數據  
        } catch (IOException e) {  
            System.out.println("[Warning!] A client has been closed.");  
            closeChannel();  
        }  
    }  
    private void closeChannel() {  
        try {  
            sk.cancel();  
            sc.close();  
        } catch (IOException e1) {  
            e1.printStackTrace();  
        }  
    }  
    private synchronized void read() throws IOException {  
        // non-blocking下不可用Readers，因為Readers不支援non-blocking  
        byte[] arr = new byte[1024];  
        ByteBuffer buf = ByteBuffer.wrap(arr);  
        int numBytes = sc.read(buf); // 讀取字符串  
        if(numBytes == -1)  
        {  
            System.out.println("[Warning!] A client has been closed.");  
            closeChannel();  
            return;  
        }  
        String str = new String(arr); // 將讀取到的byte內容轉為字符串型態  
        if ((str != null) && !str.equals(" ")) {  
            process(str); // 邏輯處理  
            System.out.println(sc.socket().getRemoteSocketAddress().toString()  
                    + " > " + str);  
            state = 1; // 改變狀態  
            sk.interestOps(SelectionKey.OP_WRITE); // 通過key改變通道註冊的事件  
            sk.selector().wakeup(); // 使一個阻塞住的selector操作立即返回  
        }  
    }  
    private void send() throws IOException  {  
        // get message from message queue  
        String str = "Your message has sent to "  
                + sc.socket().getLocalSocketAddress().toString() + "\r\n";  
        ByteBuffer buf = ByteBuffer.wrap(str.getBytes()); // wrap自動把buf的position設為0，所以不需要再flip()  
        while (buf.hasRemaining()) {  
            sc.write(buf); // 回傳給client回應字符串，發送buf的position位置 到limit位置為止之間的內容  
        }  
        state = 0; // 改變狀態  
        sk.interestOps(SelectionKey.OP_READ); // 通過key改變通道註冊的事件  
        sk.selector().wakeup(); // 使一個阻塞住的selector操作立即返回  
    }  
    void process(String str) {  
        // do process(decode, logically process, encode)..  
        // ..  
    }  
}

最后是主程序代码

[Main.java]

package server;  
import java.io.IOException;  
public class Main {  
    public static void main(String[] args) {  
        // TODO Auto-generated method stub  
        try {  
            TCPReactor reactor = new TCPReactor(1333);  
            reactor.run();  
        } catch (IOException e) {  
            // TODO Auto-generated catch block  
            e.printStackTrace();  
        }  
    }  
}

下面附上客戶端代碼：

[Client.java]

package main.pkg;  
import java.io.BufferedReader;  
import java.io.IOException;  
import java.io.InputStreamReader;  
import java.io.PrintWriter;  
import java.net.Socket;  
import java.net.UnknownHostException;  
public class Client {  
    /** 
     * @param args 
     */  
    public static void main(String[] args) {  
        // TODO Auto-generated method stub  
        String hostname=args[0];  
        int port = Integer.parseInt(args[1]);  
        //String hostname="127.0.0.1";  
        //int port=1333;  
        System.out.println("Connecting to "+ hostname +":"+port);  
        try {  
            Socket client = new Socket(hostname, port); // 連接至目的地  
            System.out.println("Connected to "+ hostname);  
            PrintWriter out = new PrintWriter(client.getOutputStream());  
            BufferedReader in = new BufferedReader(new InputStreamReader(client.getInputStream()));  
            BufferedReader stdIn = new BufferedReader(new InputStreamReader(System.in));  
            String input;  
            while((input=stdIn.readLine()) != null) { // 讀取輸入  
                out.println(input); // 發送輸入的字符串  
                out.flush(); // 強制將緩衝區內的數據輸出  
                if(input.equals("exit"))  
                {  
                    break;  
                }  
                System.out.println("server: "+in.readLine());  
            }  
            client.close();  
            System.out.println("client stop.");  
        } catch (UnknownHostException e) {  
            // TODO Auto-generated catch block  
            System.err.println("Don't know about host: " + hostname);  
        } catch (IOException e) {  
            // TODO Auto-generated catch block  
            System.err.println("Couldn't get I/O for the socket connection");  
        }  
    }  
}

到此这篇关于RabbitMQ消息队列中多路复用Channel信道详解的文章就介绍到这了,更多相关RabbitMQ多路复用Channel内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

RabbitMQ消息队列中多路复用Channel信道详解

什么叫消息队列

RabbitMQ 基本概念

RabbitMQ 内部结构

Channel

代码实现

Reactor模式的组成角色

代码实现

您可能感兴趣的文章: