Netty 快速入门系列 - Chapter 8 数据包协议【第二十讲】解决方案-Pipeline用例代码

Server 端
Upstream1ChannelHandler： [b]ChannelBuffer  转换成 String ，通过 [b]sendUpstream 让next handler 处理。[/b][/b]
ChannelBuffer channelBuffer = (ChannelBuffer) e.getMessage();
String message = new String(channelBuffer.array(), "UTF-8");
ctx.sendUpstream(upstreamMessageEvent);
package com.john.netty.learn.ch10.pipelines;

import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;
import org.jboss.netty.channel.UpstreamMessageEvent;

public class Upstream1ChannelHandler extends SimpleChannelHandler {

@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {

System.out.println("Upstream1ChannelHandler before message : " + e.getMessage().getClass().getSimpleName());

ChannelBuffer channelBuffer = (ChannelBuffer) e.getMessage();

String message = new String(channelBuffer.array(), "UTF-8");

UpstreamMessageEvent upstreamMessageEvent = new UpstreamMessageEvent(e.getChannel(), message,
e.getRemoteAddress());

System.out.println("Upstream1ChannelHandler send message : " + message);

ctx.sendUpstream(upstreamMessageEvent);
}
}

Upstream2ChannelHandler ： 直接获取String 不再是 ChannelBuffer 了。
packagecom.john.netty.learn.ch10.pipelines;

importorg.jboss.netty.channel.ChannelHandlerContext;
importorg.jboss.netty.channel.MessageEvent;
importorg.jboss.netty.channel.SimpleChannelHandler;

publicclassUpstream2ChannelHandler extendsSimpleChannelHandler {

@Override
publicvoidmessageReceived(ChannelHandlerContextctx, MessageEvent e)throwsException {

Stringmessage= (String)e.getMessage();

System.out.println("message : "+ message+ " in Upstream2ChannelHandler");

super.messageReceived(ctx,e);
}
}

PipelinePrincipleServer：package com.john.netty.learn.ch10.pipelines;

import java.net.InetSocketAddress;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;

public class PipelinePrincipleServer {

public PipelinePrincipleServer() {

}

public void bind(int port) {

ServerBootstrap serverBootstrap = new ServerBootstrap();

serverBootstrap.setFactory(new NioServerSocketChannelFactory(boss, workers));

serverBootstrap.setPipelineFactory(new ChannelPipelineFactory() {

@Override
public ChannelPipeline getPipeline() throws Exception {

ChannelPipeline channelPipeline = Channels.pipeline();

channelPipeline.addLast("handler1", new Upstream1ChannelHandler());
channelPipeline.addLast("handler2", new Upstream2ChannelHandler());

return channelPipeline;
}
});

serverBootstrap.bind(new InetSocketAddress(port));
}

public static void main(String[] args) {

new PipelinePrincipleServer().bind(23);

}

private Executor boss = Executors.newCachedThreadPool();

private Executor workers = Executors.newCachedThreadPool();
}


ClientSocket package com.john.netty.learn.ch10.pipelines;

import java.io.IOException;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.Scanner;

public class ClientSocket {

public static void main(String[] args) throws UnknownHostException, IOException {

Socket socket = new Socket("127.0.0.1", 23);

System.out.println("Please input:");

Scanner scanner = new Scanner(System.in);

socket.getOutputStream().write(scanner.nextLine().getBytes("utf-8"));

scanner.close();

socket.close();
}
}

所有源码下载 ：https://download.csdn.net/download/netcobol/10308871