标签：
在Netty中提供了ChannelGroup接口，该接口继承Set接口，因此可以通过ChannelGroup可管理服务器端所有的连接的Channel，然后对所有的连接Channel广播消息。
Server端：
public class BroadCastServer { public static void run(int port) { EventLoopGroup boss = new NioEventLoopGroup(); EventLoopGroup worker = new NioEventLoopGroup(); try { ServerBootstrap bootstrap = new ServerBootstrap(); bootstrap.group(boss, worker) .channel(NioServerSocketChannel.class) // 设置Channel Type .option(ChannelOption.SO_BACKLOG, 1024) // 设置Channel属性 .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4)); pipeline.addLast("frameEncoder", new LengthFieldPrepender(4)); pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8)); pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8)); pipeline.addLast(new BroadCastChannelHandler()); } }); ChannelFuture channelFuture = bootstrap.bind(port).sync(); if (channelFuture.isDone()) { System.out.println(String.format("server bind port %s sucess", port)); } Channel channel = channelFuture.channel(); /**CloseFuture异步方式关闭*/ channel.closeFuture().sync(); } catch (InterruptedException e) { e.printStackTrace(); } finally { boss.shutdownGracefully(); worker.shutdownGracefully(); } } public static void main(String []args) { BroadCastServer.run(8080); } } public class BroadCastChannelHandler extends ChannelInboundHandlerAdapter { private static final Gson GSON = new GsonBuilder().create(); private static final SimpleDateFormat SDF = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); private static final AtomicInteger response = new AtomicInteger(); @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { Channel ch = ctx.channel(); if (ChannelGroups.size() > 0) { Message msg = new Message(ch.remoteAddress().toString().substring(1), SDF.format(new Date())); ChannelGroups.broadcast(GSON.toJson(msg), new ChannelMatchers()); } ChannelGroups.add(ch); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { Channel ch = ctx.channel(); if (ChannelGroups.contains(ch) && String.valueOf(msg).equals("welcome")) { System.out.println(String.format("receive [%s] from [%s] at [%s]", String.valueOf(msg) , ch.remoteAddress().toString().substring(1), SDF.format(new Date()))); response.incrementAndGet(); } synchronized (response) { System.out.println(response.get() + "\t" + ChannelGroups.size()); if (response.get() == ChannelGroups.size() - 1) { System.out.println("server close all connected channel"); ChannelGroups.disconnect(); response.set(0); } } } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ctx.flush(); } @Override public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { ctx.close(); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { cause.printStackTrace(); ChannelGroups.discard(ctx.channel()); response.decrementAndGet(); } public static class ChannelMatchers implements ChannelMatcher { @Override public boolean matches(Channel channel) { return true; } } }
服务器端收到所有连接客户端对广播消息的响应后，服务器端主动关闭已连接的Channel
客户端：
public class Client {
private static final String host = "127.0.0.1";
private static final int port = 8080;
private static final ExecutorService es = Executors.newFixedThreadPool(5);
public static void start() {
for (int i = 0; i < 5; i++) {
es.execute(new Task());
}
es.shutdown();
}
public static class Task implements Runnable {
@Override
public void run() {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
pipeline.addLast(new SimpleClientChannelHandler());
}
});
ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
if (channelFuture.isSuccess()) {
System.out.println(String.format("connect server(%s:%s) sucess", host, port));
}
channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
group.shutdownGracefully();
}
}
}
public static void main(String []args) {
Client.start();
}
}
public class SimpleClientChannelHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
Channel channel = ctx.channel();
System.out.println(String.format("client(%s) receive message [%s]", channel.localAddress().toString().substring(1),
String.valueOf(msg)));
System.out.println();
ctx.writeAndFlush(String.valueOf("welcome"));
}
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
ctx.disconnect(ctx.newPromise());
ctx.close();
System.out.println(String.format("client(%s) close sucess", ctx.channel().localAddress().toString().substring(1)));
}
}
本文使用ChannelGroups辅助类管理服务器端已连接的Channel，代码实现如下:
public class ChannelGroups {
private static final ChannelGroup CHANNEL_GROUP = new DefaultChannelGroup("ChannelGroups", GlobalEventExecutor.INSTANCE);
public static void add(Channel channel) {
CHANNEL_GROUP.add(channel);
}
public static ChannelGroupFuture broadcast(Object msg) {
return CHANNEL_GROUP.writeAndFlush(msg);
}
public static ChannelGroupFuture broadcast(Object msg, ChannelMatcher matcher) {
return CHANNEL_GROUP.writeAndFlush(msg, matcher);
}
public static ChannelGroup flush() {
return CHANNEL_GROUP.flush();
}
public static boolean discard(Channel channel) {
return CHANNEL_GROUP.remove(channel);
}
public static ChannelGroupFuture disconnect() {
return CHANNEL_GROUP.disconnect();
}
public static ChannelGroupFuture disconnect(ChannelMatcher matcher) {
return CHANNEL_GROUP.disconnect(matcher);
}
public static boolean contains(Channel channel) {
return CHANNEL_GROUP.contains(channel);
}
public static int size() {
return CHANNEL_GROUP.size();
}
}
标签：
原文地址：http://www.cnblogs.com/hanfight/p/4852496.html
