Mina、Netty、Twisted一块儿学（七）：发布/订阅（Publish/Subscribe）

消息传递有不少种方式，请求/响应（Request/Reply）是最经常使用的。在前面的博文的例子中，不少都是采用请求/响应的方式，当服务器接收到消息后，会当即write回写一条消息到客户端。HTTP协议也是基于请求/响应的方式。

可是请求/响应并不能知足全部的消息传递的需求，有些需求可能须要服务端主动推送消息到客户端，而不是被动的等待请求后再给出响应。

发布/订阅（Publish/Subscribe）是一种服务器主动发送消息到客户端的消息传递方式。订阅者Subscriber链接到服务器客户端后，至关于开始订阅发布者Publisher发布的消息，当发布者发布了一条消息后，全部订阅者都会接收到这条消息。

网络聊天室通常就是基于发布/订阅模式来实现。例如加入一个QQ群，就至关于订阅了这个群的全部消息，当有新的消息，服务器会主动将消息发送给全部的客户端。只不过聊天室里的全部人既是发布者又是订阅者。

下面分别用MINA、Netty、Twisted分别实现简单的发布/订阅模式的服务器程序，链接到服务器的全部客户端都是订阅者，当发布者发布一条消息后，服务器会将消息转发给全部客户端。

MINA：

在MINA中，经过IoService的getManagedSessions()方法能够获取这个IoService当前管理的全部IoSession，即全部链接到服务器的客户端集合。当服务器接收到发布者发布的消息后，能够经过IoService的getManagedSessions()方法获取到全部客户端对应的IoSession并将消息发送到这些客户端。

public class TcpServer {

    public static void main(String[] args) throws IOException {
        IoAcceptor acceptor = new NioSocketAcceptor();

        acceptor.getFilterChain().addLast("codec",
                new ProtocolCodecFilter(new TextLineCodecFactory(Charset.forName("UTF-8"), "\r\n", "\r\n")));

        acceptor.setHandler(new TcpServerHandle());
        acceptor.bind(new InetSocketAddress(8080));
    }

}

class TcpServerHandle extends IoHandlerAdapter {

    @Override
    public void exceptionCaught(IoSession session, Throwable cause)
            throws Exception {
        cause.printStackTrace();
    }

    @Override
    public void messageReceived(IoSession session, Object message)
            throws Exception {
        
        // 获取全部正在链接的IoSession
        Collection<IoSession> sessions = session.getService().getManagedSessions().values();

        // 将消息写到全部IoSession
        IoUtil.broadcast(message, sessions);
    }
}

Netty：

Netty提供了ChannelGroup来用于保存Channel组，ChannelGroup是一个线程安全的Channel集合，它提供了一些列Channel批量操做。当一个TCP链接关闭后，对应的Channel会自动从ChannelGroup移除，因此不用手动去移除关闭的Channel。

Netty文档关于ChannelGroup的解释：

A thread-safe Set that contains open Channels and provides various bulk operations on them. Using ChannelGroup, you can categorize Channels into a meaningful group (e.g. on a per-service or per-state basis.) A closed Channel is automatically removed from the collection, so that you don't need to worry about the life cycle of the added Channel. A Channel can belong to more than one ChannelGroup.

当有新的客户端链接到服务器，将对应的Channel加入到一个ChannelGroup中，当发布者发布消息时，服务器能够将消息经过ChannelGroup写入到全部客户端。

public class TcpServer {

    public static void main(String[] args) throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new LineBasedFrameDecoder(80));
                            pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
                            pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
                            pipeline.addLast(new TcpServerHandler());
                        }
                    });
            ChannelFuture f = b.bind(8080).sync();
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
}

class TcpServerHandler extends ChannelInboundHandlerAdapter {

    // ChannelGroup用于保存全部链接的客户端，注意要用static来保证只有一个ChannelGroup实例，不然每new一个TcpServerHandler都会建立一个ChannelGroup
    private static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        channels.add(ctx.channel()); // 将新的链接加入到ChannelGroup，当链接断开ChannelGroup会自动移除对应的Channel
    }
    
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        channels.writeAndFlush(msg + "\r\n"); // 接收到消息后，将消息发送到ChannelGroup中的全部客户端
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // cause.printStackTrace();  暂时把异常打印注释掉，由于PublishClient发布一条消息后会当即断开链接，而服务器也会向PublishClient发送消息，因此会抛出异常
        ctx.close();
    }
}

Twisted：

在Twisted中，全局的数据通常会放在Factory，而每一个链接相关的数据会放在Protocol中。因此这里能够在Factory中加入一个属性，来存放Protocol集合，表示全部链接服务器的客户端。当有新的客户端链接到服务器时，将对应的Protocol实例放入集合，当链接断开，将对应的Protocol从集合中移除。当服务器接收到发布者发布的消息后，遍历全部客户端并发送消息。

# -*- coding:utf-8 –*-

from twisted.protocols.basic import LineOnlyReceiver
from twisted.internet.protocol import Factory
from twisted.internet import reactor

class TcpServerHandle(LineOnlyReceiver): 

    def __init__(self, factory):
        self.factory = factory

    def connectionMade(self):
        self.factory.clients.add(self) # 新链接添加链接对应的Protocol实例到clients

    def connectionLost(self, reason):
        self.factory.clients.remove(self) # 链接断开移除链接对应的Protocol实例

    def lineReceived(self, line):
        # 遍历全部的链接，发送数据
        for c in self.factory.clients:
            c.sendLine(line)

class TcpServerFactory(Factory):
    def __init__(self):
        self.clients = set() # set集合用于保存全部链接到服务器的客户端

    def buildProtocol(self, addr):
        return TcpServerHandle(self)

reactor.listenTCP(8080, TcpServerFactory())
reactor.run()

下面分别是两个客户端程序，一个是用于发布消息的客户端，一个是订阅消息的客户端。

发布消息的客户端很简单，就是向服务器write一条消息便可：

public class PublishClient {

    public static void main(String[] args) throws IOException {

        Socket socket = null;
        OutputStream out = null;

        try {

            socket = new Socket("localhost", 8080);
            out = socket.getOutputStream();
            out.write("Hello\r\n".getBytes()); // 发布信息到服务器
            out.flush();

        } finally {
            // 关闭链接
            out.close();
            socket.close();
        }
    }
}

订阅消息的客户端链接到服务器后，会阻塞等待接收服务器发送的发布消息：

public class SubscribeClient {

    public static void main(String[] args) throws IOException {

        Socket socket = null;
        BufferedReader in = null;

        try {

            socket = new Socket("localhost", 8080);
            in = new BufferedReader(new InputStreamReader(socket.getInputStream()));

            while (true) {
                String line = in.readLine(); // 阻塞等待服务器发布的消息
                System.out.println(line);
            }

        } finally {
            // 关闭链接
            in.close();
            socket.close();
        }
    }
}

分别针对MINA、Netty、Twisted服务器进行测试：

一、测试时首先开启服务器；
二、而后再运行订阅消息的客户端SubscribeClient，SubscribeClient能够开启多个；
三、最后运行发布消息的客户端PublishClient，能够屡次运行查看全部SubscribeClient的输出结果。

运行结果能够发现，当运行发布消息的客户端PublishClient发布一条消息到服务器时，服务器会主动将这条消息转发给全部的TCP链接，全部的订阅消息的客户端SubscribeClient都会接收到这条消息并打印出来。

MINA、Netty、Twisted一块儿学系列

MINA、Netty、Twisted一块儿学（一）：实现简单的TCP服务器

MINA、Netty、Twisted一块儿学（二）：TCP消息边界问题及按行分割消息

MINA、Netty、Twisted一块儿学（三）：TCP消息固定大小的前缀(Header)

MINA、Netty、Twisted一块儿学（四）：定制本身的协议

MINA、Netty、Twisted一块儿学（五）：整合protobuf

MINA、Netty、Twisted一块儿学（六）：session

MINA、Netty、Twisted一块儿学（七）：发布/订阅（Publish/Subscribe）

MINA、Netty、Twisted一块儿学（八）：HTTP服务器

MINA、Netty、Twisted一块儿学（九）：异步IO和回调函数

MINA、Netty、Twisted一块儿学（十）：线程模型

MINA、Netty、Twisted一块儿学（十一）：SSL/TLS

MINA、Netty、Twisted一块儿学（十二）：HTTPS

源码

https://github.com/wucao/mina-netty-twisted