相关推荐:Mina、Netty、Twisted一起学(六):session

开发过Web应用的同学应该都会使用session。由于HTTP协议本身是无状态的,所以一个客户端多次访问这个web应用的多个页面,服务器无法判断多次访问的客户端是否是同一个客户端。有了session就可以设置一些和客户端相关的属性,用于保持这种

消息传递有很多种方式,请求/响应(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 LineOnlyReceiverfrom twisted.internet.protocol import Factoryfrom twisted.internet import reactorclass 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服务器进行测试:1、测试时首先开启服务器;2、然后再运行订阅消息的客户端SubscribeClient,SubscribeClient可以开启多个;3、最后运行发布消息的客户端PublishClient,可以多次运行查看所有SubscribeClient的输出结果。运行结果可以发现,当运行发布消息的客户端PublishClient发布一条消息到服务器时,服务器会主动将这条消息转发给所有的TCP连接,所有的订阅消息的客户端SubscribeClient都会接收到这条消息并打印出来。

相关推荐:Mina、Netty、Twisted一起学(十):线程模型

要想开发一个高性能的TCP服务器,熟悉所使用框架的线程模型非常重要。MINA、Netty、Twisted本身都是高性能的网络框架,如果再搭配上高效率的代码,才能实现一个高大上的服务器。但是如果不了解它们的线程模型,就很难写出高性能的代码。

快照源:http://www.cnblogs.com/wucao/p/3985407.html