Netty如何寫監(jiān)視器

這一節(jié)我們學(xué)習(xí)監(jiān)視器的編寫：EventLogMonitor ，也就是用來接收事件的程序，用來代替 netcat 。EventLogMonitor 做下面事情：

• 接收 LogEventBroadcaster 廣播的 UDP DatagramPacket
• 解碼 LogEvent 消息
• 輸出 LogEvent 消息

和之前一樣,將實(shí)現(xiàn)自定義 ChannelHandler 的邏輯。圖13.4描述了LogEventMonitor 的 ChannelPipeline 并表明了 LogEvent 的流經(jīng)情況。

Figure 13.4 LogEventMonitor

圖中顯示我們的兩個(gè)自定義 ChannelHandlers,LogEventDecoder 和 LogEventHandler。首先是負(fù)責(zé)將網(wǎng)絡(luò)上接收到的 DatagramPacket 解碼到 LogEvent 消息。清單13.6顯示了實(shí)現(xiàn)。

Listing 13.6 LogEventDecoder

public class LogEventDecoder extends MessageToMessageDecoder<DatagramPacket> {
    @Override
    protected void decode(ChannelHandlerContext ctx, DatagramPacket datagramPacket, List<Object> out) throws Exception {
        ByteBuf data = datagramPacket.content(); //1
        int i = data.indexOf(0, data.readableBytes(), LogEvent.SEPARATOR);  //2
        String filename = data.slice(0, i).toString(CharsetUtil.UTF_8);  //3
        String logMsg =  data.slice(i + 1, data.readableBytes()).toString(CharsetUtil.UTF_8);  //4

        LogEvent event = new LogEvent(datagramPacket.recipient(), System.currentTimeMillis(),
                filename,logMsg); //5
        out.add(event);
    }
}

1. 獲取 DatagramPacket 中數(shù)據(jù)的引用
2. 獲取 SEPARATOR 的索引
3. 從數(shù)據(jù)中讀取文件名
4. 讀取數(shù)據(jù)中的日志消息
5. 構(gòu)造新的 LogEvent 對(duì)象并將其添加到列表中

第二個(gè) ChannelHandler 將執(zhí)行一些首先創(chuàng)建的 LogEvent 消息。在這種情況下,我們只會(huì)寫入 system.out。在真實(shí)的應(yīng)用程序可能用到一個(gè)單獨(dú)的日志文件或放到數(shù)據(jù)庫(kù)。

下面的清單顯示了 LogEventHandler。

Listing 13.7 LogEventHandler

public class LogEventHandler extends SimpleChannelInboundHandler<LogEvent> { //1

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace(); //2
        ctx.close();
    }

    @Override
    public void channelRead0(ChannelHandlerContext channelHandlerContext, LogEvent event) throws Exception {
        StringBuilder builder = new StringBuilder(); //3
        builder.append(event.getReceivedTimestamp());
        builder.append(" [");
        builder.append(event.getSource().toString());
        builder.append("] [");
        builder.append(event.getLogfile());
        builder.append("] : ");
        builder.append(event.getMsg());

        System.out.println(builder.toString()); //4
    }
}

1. 繼承 SimpleChannelInboundHandler 用于處理 LogEvent 消息
2. 在異常時(shí)，輸出消息并關(guān)閉 channel
3. 建立一個(gè) StringBuilder 并構(gòu)建輸出
4. 打印出 LogEvent 的數(shù)據(jù)

LogEventHandler 打印出 LogEvent 的一個(gè)易讀的格式,包括以下:

• 收到時(shí)間戳以毫秒為單位
• 發(fā)送方的 InetSocketAddress,包括IP地址和端口
• LogEvent 生成絕對(duì)文件名
• 實(shí)際的日志消息,代表在日志文件中一行

現(xiàn)在我們需要安裝處理程序到 ChannelPipeline ，如圖13.4所示。下一個(gè)清單顯示了這是如何實(shí)現(xiàn) LogEventMonitor 類的一部分。

Listing 13.8 LogEventMonitor

public class LogEventMonitor {

    private final Bootstrap bootstrap;
    private final EventLoopGroup group;
    public LogEventMonitor(InetSocketAddress address) {
        group = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(group)  //1
                .channel(NioDatagramChannel.class)
                .option(ChannelOption.SO_BROADCAST, true)
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel channel) throws Exception {
                        ChannelPipeline pipeline = channel.pipeline();
                        pipeline.addLast(new LogEventDecoder());  //2
                        pipeline.addLast(new LogEventHandler());
                    }
                }).localAddress(address);

    }

    public Channel bind() {
        return bootstrap.bind().syncUninterruptibly().channel();  //3
    }

    public void stop() {
        group.shutdownGracefully();
    }

    public static void main(String[] args) throws Exception {
        if (args.length != 1) {
            throw new IllegalArgumentException("Usage: LogEventMonitor <port>");
        }
        LogEventMonitor monitor = new LogEventMonitor(new InetSocketAddress(Integer.parseInt(args[0])));  //4
        try {
            Channel channel = monitor.bind();
            System.out.println("LogEventMonitor running");

            channel.closeFuture().await();
        } finally {
            monitor.stop();
        }
    }
}

1. 引導(dǎo) NioDatagramChannel。設(shè)置 SO_BROADCAST socket 選項(xiàng)。
2. 添加 ChannelHandler 到 ChannelPipeline
3. 綁定的通道。注意,在使用 DatagramChannel 是沒有連接，因?yàn)檫@些 無連接
4. 構(gòu)建一個(gè)新的 LogEventMonitor