Netty利用ChannelGroup广播消息

在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();
}
}