关于netty捕获应答报文的问题
程序员文章站
2022-03-02 15:25:42
...
客户端代码如下：
package com.lt.netty.cilent;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
public class EchoCilent {
private final String host;
private final int port;
public EchoCilent(String host,int port){
this.port = port;
this.host = host;
}
public static void main(String[] args) throws Exception {
final String host = "localhost";
final int port = 8888;
new EchoCilent(host, port).start();
}
public void start() throws Exception{
//多线程事件循环器
//EventLoopGroup group = new OioEventLoopGroup();
EventLoopGroup group = new NioEventLoopGroup();
try {
//创建配置管理器
Bootstrap b = new Bootstrap();
b.group(group)//配置工作组
.channel(NioSocketChannel.class)//配置通道
.remoteAddress(host, port)//连接地址
.handler(new ChannelInitializer<SocketChannel>() {//初始化通道
@Override
public void initChannel(SocketChannel ch){
ch.pipeline().addLast(new EchoCilentHandler());//添加处理类
}
});
ChannelFuture f = b.connect().sync();//同步等待关闭配置管理器连接
//通过ChannelFuture获取处理类对象调用其方法获取应答报文
EchoCilentHandler handler = (EchoCilentHandler)f.channel().pipeline().last();
System.out.println("handler.getName()="+handler.getFactorial());
f.channel().closeFuture().sync();//同步等待关闭通道
} finally {
group.shutdownGracefully().sync();//同步等待关闭循环器
}
}
}客户端处理类代码如下：
package com.lt.netty.cilent;
import java.io.UnsupportedEncodingException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
@Sharable
public class EchoCilentHandler extends SimpleChannelInboundHandler<ByteBuf>{
//声明队列用来存取应答报文
final BlockingQueue<String> answer = new LinkedBlockingQueue<String>();
public String getFactorial() {
boolean interrupted = false;
try {
for (;;) {
try {
//获取队列中的消息并返回
return answer.take();
} catch (InterruptedException ignore) {
interrupted = true;
}
}
} finally {
if (interrupted) {
Thread.currentThread().interrupt();
}
}
}
//当与服务端连接上时
@Override
public void channelActive(ChannelHandlerContext ctx) throws UnsupportedEncodingException{
System.out.println("channelActive is ok");
String msg = "<?xml version=\"1.0\" encoding=\"UTF-8\"?><transaction><header><msg><rcvAppCd>SMS</rcvAppCd><sndAppCd>ESB</sndAppCd></msg></header><body><request><MblPhNo>18713341990</MblPhNo></request></body></transaction>";
String len = String.format("%08d", msg.getBytes("utf-8").length);
byte[] bt = (len+msg).getBytes("utf-8");
ByteBuf buf = ctx.alloc().buffer();
buf.writeBytes(bt);
//写入并刷新缓冲区
ctx.writeAndFlush(buf);
}
//收到服务端并开始读取时
@Override
public void channelRead0(ChannelHandlerContext ctx,ByteBuf msg){
ByteBuf in = (ByteBuf) msg;System.out.println("cilent read:"+in.toString(io.netty.util.CharsetUtil.UTF_8));
//将应答报文放入队列中
answer.offer(in.toString(io.netty.util.CharsetUtil.UTF_8));
}
//生命周期内发生异常
@Override
public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause){
cause.printStackTrace();
ctx.close();
}
}
服务端代码：
package com.lt.netty.cilent;
import java.net.InetSocketAddress;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
public class EchoServer {
private final int port;
public EchoServer(int port){
this.port = port;
}
public static void main(String[] args) throws Exception {
int port = 8888;
new EchoServer(port).start();
}
public void start() throws Exception{
//创建非阻塞多线程事件循环器
NioEventLoopGroup group = new NioEventLoopGroup();
try {
//创建服务端配置管理器
ServerBootstrap b = new ServerBootstrap();
//配置信息
b.group(group).channel(NioServerSocketChannel.class)
.localAddress(new InetSocketAddress(port))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception{
ch.pipeline().addLast(new EchoServerHandler());
}
});
//服务端配置器绑定连接
ChannelFuture f = b.bind().sync();
System.out.println(EchoServer.class.getName()+"started and listens on"
+f.channel().localAddress());
//关闭通道
f.channel().closeFuture().sync();
} finally {
group.shutdownGracefully();
}
}
}服务端处理类代码：
package com.lt.netty.cilent;
import java.io.UnsupportedEncodingException;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;
@Sharable
public class EchoServerHandler extends ChannelInboundHandlerAdapter{
@Override
public void channelRead(ChannelHandlerContext ctx,Object msg) throws UnsupportedEncodingException{
ByteBuf in = (ByteBuf) msg;
System.out.println("server received:"+in.toString(CharsetUtil.UTF_8));
String str = "<?xml version=\"1.0\" encoding=\"UTF-8\"?><transaction><header><status><desc>实时短信服务请求成功！</desc><retCd>000000</retCd></status><msg><sndAppCd>SMS</sndAppCd><rcvAppCd>ESB</rcvAppCd></msg></header><body><MblPhNo>18713341990</MblPhNo><StCD>000000</StCD></body></transaction>";
String len = String.format("%08d", str.getBytes("utf-8").length);
byte[] bt = (len+str).getBytes("utf-8");
ByteBuf buf = ctx.alloc().buffer();
ctx.writeAndFlush(buf.writeBytes(bt));
}
//当应答消息被读取完成时
@Override
public void channelReadComplete(ChannelHandlerContext ctx){
ctx.writeAndFlush(Unpooled.EMPTY_BUFFER)
.addListener(ChannelFutureListener.CLOSE);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause){
cause.printStackTrace();
ctx.close();
}
}
下一篇： 双向关联多对多