13.4 编写广播器

    我们要做的是广播一个DatagramPacket日志条目，如下图所示：

    上图显示我们有一个从日志条路到DatagramPacket一对一的关系。如同所有的基于Netty的应用程序一样，它由一个或多个ChannelHandler和一些实体对象绑定，用于引导该应用程序。首先让我们来看看LogEventBroadcaster的ChannelPipeline以及作为数据载体的LogEvent的流向，看下图：

    上图显示，LogEventBroadcaster使用LogEvent消息并将消息写入本地Channel，所有的信息封装在LogEvent消息中，这些消息被传到ChannelPipeline中。流进ChannelPipeline的LogEvent消息被编码成DatagramPacket消息，最后通过UDP广播到远程对等通道。

    这可以归结为有一个自定义的ChannelHandler，从LogEvent消息编程成DatagramPacket消息。回忆我们在第七章讲解的编解码器，我们定义个LogEventEncoder，代码如下：

[java] view plaincopy

1. package netty.in.action.udp;
3. import io.netty.buffer.ByteBuf;
4. import io.netty.channel.ChannelHandlerContext;
5. import io.netty.channel.socket.DatagramPacket;
6. import io.netty.handler.codec.MessageToMessageEncoder;
7. import io.netty.util.CharsetUtil;
9. import java.net.InetSocketAddress;
10. import java.util.List;
12. public class LogEventEncoder extends MessageToMessageEncoder<LogEvent> {
14. private final InetSocketAddress remoteAddress;
16. public LogEventEncoder(InetSocketAddress remoteAddress){
17. this.remoteAddress = remoteAddress;
18. }
20. @Override
21. protected void encode(ChannelHandlerContext ctx, LogEvent msg, List<Object> out)
22. throws Exception {
23. ByteBuf buf = ctx.alloc().buffer();
24. buf.writeBytes(msg.getLogfile().getBytes(CharsetUtil.UTF_8));
25. buf.writeByte(LogEvent.SEPARATOR);
26. buf.writeBytes(msg.getMsg().getBytes(CharsetUtil.UTF_8));
27. out.add(new DatagramPacket(buf, remoteAddress));
28. }

30. }

    下面我们再编写一个广播器：

[java] view plaincopy

1. package netty.in.action.udp;
3. import io.netty.bootstrap.Bootstrap;
4. import io.netty.channel.Channel;
5. import io.netty.channel.ChannelOption;
6. import io.netty.channel.EventLoopGroup;
7. import io.netty.channel.nio.NioEventLoopGroup;
8. import io.netty.channel.socket.nio.NioDatagramChannel;
10. import java.io.File;
11. import java.io.IOException;
12. import java.io.RandomAccessFile;
13. import java.net.InetSocketAddress;
15. public class LogEventBroadcaster {
17. private final EventLoopGroup group;
18. private final Bootstrap bootstrap;
19. private final File file;
21. public LogEventBroadcaster(InetSocketAddress address, File file) {
22. group = new NioEventLoopGroup();
23. bootstrap = new Bootstrap();
24. bootstrap.group(group).channel(NioDatagramChannel.class)
25. .option(ChannelOption.SO_BROADCAST, true)
26. .handler(new LogEventEncoder(address));
27. this.file = file;
28. }
30. public void run() throws IOException {
31. Channel ch = bootstrap.bind(0).syncUninterruptibly().channel();
32. long pointer = 0;
33. for (;;) {
34. long len = file.length();
35. if (len < pointer) {
36. pointer = len;
37. } else {
38. RandomAccessFile raf = new RandomAccessFile(file, "r");
39. raf.seek(pointer);
40. String line;
41. while ((line = raf.readLine()) != null) {
42. ch.write(new LogEvent(null, -1, file.getAbsolutePath(), line));
43. }
44. ch.flush();
45. pointer = raf.getFilePointer();
46. raf.close();
47. }
48. try {
49. Thread.sleep(1000);
50. } catch (InterruptedException e) {
51. Thread.interrupted();
52. break;
53. }
54. }
55. }
57. public void stop() {
58. group.shutdownGracefully();
59. }
61. public static void main(String[] args) throws Exception {
62. int port = 4096;
63. String path = System.getProperty("user.dir") + "/log.txt";
64. LogEventBroadcaster broadcaster = new LogEventBroadcaster(new InetSocketAddress(
65. "255.255.255.255", port), new File(path));
66. try {
67. broadcaster.run();
68. } finally {
69. broadcaster.stop();
70. }
71. }
73. }