Java：Netty与Thrift整合的例子

痛定思痛。 2023-06-25 15:29 14阅读 0赞

Netty官方没的提供与Thrift整合的API，下面就提供一个整合的简单例子。

Maven:

> <dependency>  
>     <groupId>org.apache.thrift</groupId>  
>     <artifactId>libthrift</artifactId>  
>     <version>0.13.0</version>  
> </dependency>  
> <dependency>  
>     <groupId>io.netty</groupId>  
>     <artifactId>netty-all</artifactId>  
>     <version>4.1.44.Final</version>  
> </dependency>

tutorial.thrift

> namespace java com.test
> 
> enum Operation \{  
>   ADD = 1,  
>   SUBTRACT = 2,  
>   MULTIPLY = 3,  
>   DIVIDE = 4  
> \}
> 
> struct Work \{  
>   1: i32 num1 = 0,  
>   2: i32 num2,  
>   3: Operation op,  
>   4: optional string comment,  
> \}
> 
> service Calculator \{  
>    i32 add(1:i32 num1, 2:i32 num2),  
>    i32 addList(1:list<Work> workList),  
> \}
> 
> service Sender \{  
>        void ping(),  
> \}

执行命令，生成Calculator.java、Sender.java、Operation.java、Work.java

> set inDir=D:/project/build/example  
> set outDir=D:/project/test
> 
> D:/project/thrift-0.13.0.exe -r --gen java -out %outDir% %inDir%/tutorial.thrift

服务器端代码：

package com.test.netty;
    
    import static com.test.netty.Constants.IDLE_TIMEOUT;
    import static com.test.netty.Constants.MAX_CONTENT_LENGTH;
    import static com.test.netty.Constants.PKG_HEAD_LEN;
    import static com.test.netty.Constants.SOCK_BACKLOG;
    import static com.test.netty.Constants.TAG_GATEWAY;
    import static com.test.netty.Constants.TAG_GATEWAY_BOSS;
    import static io.netty.channel.ChannelOption.SO_BACKLOG;
    
    import org.apache.thrift.TMultiplexedProcessor;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import com.test.Calculator;
    import com.test.Sender;
    import com.test.main.CalculatorHandler;
    import com.test.main.SenderHandler;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
    import io.netty.handler.timeout.IdleStateHandler;
    
    public class BackendServer {
    
    	private static final Logger LOG = LoggerFactory.getLogger(BackendServer.class);
    
    	private final int listenPort;
    
    	private final ServerBootstrap serverBootstrap;
    
    	private Channel listenChannel;
    
    	public BackendServer(int listenPort) {
    		TMultiplexedProcessor processor = new TMultiplexedProcessor();
    
    		processor.registerProcessor("Calculator", new Calculator.Processor<CalculatorHandler>(new CalculatorHandler()));
    		processor.registerProcessor("Sender", new Sender.Processor<SenderHandler>(new SenderHandler()));
    
    		this.listenPort = listenPort;
    
    		EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    		EventLoopGroup workerGroup = new NioEventLoopGroup();
    
    		serverBootstrap = new ServerBootstrap();
    
    		serverBootstrap.group(bossGroup, workerGroup);
    		serverBootstrap.channel(NioServerSocketChannel.class);
    		serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
    
    			protected void initChannel(SocketChannel ch) throws Exception {
    				ChannelPipeline pipeline = ch.pipeline();
    
    				pipeline.addLast("idleStateHandler",
    						new IdleStateHandler(IDLE_TIMEOUT, IDLE_TIMEOUT, 0));
    
    				pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(MAX_CONTENT_LENGTH, 0, PKG_HEAD_LEN));
    
    				pipeline.addLast("gatewayHandler", new ThriftHandler(new ThriftExecutor(processor)));
    			}
    		});
    
    		serverBootstrap.option(SO_BACKLOG, SOCK_BACKLOG);
    
    		LOG.info("initiate: listenPort=" + listenPort);
    	}
    
    	public boolean startListen() {
    		try {
    			listenChannel = serverBootstrap.bind(listenPort).sync().channel();
    
    			LOG.info("finish to startListen. listenPort=" + listenPort);
    
    			return true;
    		} catch (Exception e) {
    			LOG.error("fail to startListen. listenPort=" + listenPort, e);
    		}
    
    		return false;
    	}
    
    	public void stopListen() {
    		try {
    			if (listenChannel != null) {
    				listenChannel.disconnect().get();
    
    				LOG.info("finish to stopListen. listenPort=" + listenPort);
    			}
    		} catch (Exception e) {
    			LOG.error("fail to stopListen. listenPort=" + listenPort, e);
    		}
    	}
    
    	public static void main(String[] args) {
    		BackendServer server = new BackendServer(9090);
    		server.startListen();
    	}
    }

package com.test.netty;
    
    public final class Constants {
    
    	public static final int PKG_HEAD_LEN = 4;
    
    	public static final int MAX_CONTENT_LENGTH = 16384000;
    
    	public static final int MAX_BUFF_SIZE = 24 * 1024;
    
    	public static final int SOCK_BACKLOG = 1024;
    
    	// unit: millisecond
    	public static final long DEFAULT_PERIOD_PER_BULK = 300;
    
    	// unit: second
    	public static final int IDLE_TIMEOUT_OF_GATEWAY = 5;
    }

package com.test.netty;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.handler.timeout.IdleState;
    import io.netty.handler.timeout.IdleStateEvent;
    
    public class ThriftHandler extends SimpleChannelInboundHandler<ByteBuf> {
    
    	private static final Logger LOG = LoggerFactory.getLogger(ThriftHandler.class);
    
    	private final ThriftExecutor executor;
    
    	public ThriftHandler(ThriftExecutor executor) {
    		this.executor = executor;
    	}
    
    	public void channelActive(ChannelHandlerContext ctx) throws Exception {
    		Channel channel = ctx.channel();
    		String channelId = channel.id().asShortText();
    
    		LOG.trace("channelActive: channelId={}", channelId);
    	}
    
    	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    		Channel channel = ctx.channel();
    		String channelId = channel.id().asShortText();
    
    		LOG.trace("channelInactive: channelId={}", channelId);
    	}
    
    	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    		LOG.error("exceptionCaught:", cause);
    	}
    
    	protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
    		executor.executeAction(ctx, msg);
    	}
    
    	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    		if (evt instanceof IdleStateEvent) {
    			IdleStateEvent e = (IdleStateEvent) evt;
    
    			if (e.state() == IdleState.READER_IDLE) {
    				ctx.channel().close();
    			} else if (e.state() == IdleState.WRITER_IDLE) {
    				ctx.channel().close();
    			}
    		}
    	}
    }

package com.test.netty;
    
    import java.nio.ByteBuffer;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    import org.apache.thrift.TMultiplexedProcessor;
    import org.apache.thrift.protocol.TCompactProtocol;
    import org.apache.thrift.transport.TByteBuffer;
    import org.apache.thrift.transport.TFramedTransport;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    
    public class ThriftExecutor {
    
    	private final ExecutorService es = Executors.newFixedThreadPool(10);
    
    	private final TMultiplexedProcessor processor;
    
    	public ThriftExecutor(TMultiplexedProcessor processor) {
    		this.processor = processor;
    	}
    
    	public void executeAction(ChannelHandlerContext ctx, ByteBuf srcBuf) {
    		int len = srcBuf.readableBytes();
    		ByteBuffer inBuf = ByteBuffer.allocate(len);
    		srcBuf.getBytes(0, inBuf);
    		inBuf.flip();
    
    		es.execute(new Runnable() {
    			public void run() {
    
    				try {
    					ByteBuffer outBuf = ByteBuffer.allocate(256);
    
    					processor.process(new TCompactProtocol(new TFramedTransport(new TByteBuffer(inBuf))),
    							new TCompactProtocol(new TFramedTransport(new TByteBuffer(outBuf))));
    
    					outBuf.flip();
    					int retLen = outBuf.remaining();
    
    					ByteBuf dstBuf = ctx.alloc().buffer(retLen).writeBytes(outBuf);
    
    					ctx.writeAndFlush(dstBuf);
    				} catch (Exception e) {
    					e.printStackTrace();
    				}
    			}
    
    		});
    	}
    }

在服务器端实现服务接口：

package com.test.main;
    
    import java.util.List;
    
    import org.apache.thrift.TException;
    
    import com.test.Calculator.Iface;
    import com.test.Work;
    
    public class CalculatorHandler implements Iface {
    
    	@Override
    	public int add(int num1, int num2) throws TException {
    		System.out.println(Thread.currentThread() + ": add");
    		return num1 + num2;
    	}
    
    	@Override
    	public int addList(List<Work> workList) throws TException {
    		System.out.println(Thread.currentThread() + ": addList");
    		return workList.size();
    	}
    }

package com.test.main;
    
    import org.apache.thrift.TException;
    
    import com.test.Sender.Iface;
    
    public class SenderHandler implements Iface {
    
    	@Override
    	public void ping() throws TException {
    		System.out.println("ping");
    	}
    }

客户端代码：

package com.test.main;
    
    import java.util.ArrayList;
    import java.util.List;
    
    import org.apache.thrift.TException;
    import org.apache.thrift.protocol.TBinaryProtocol;
    import org.apache.thrift.protocol.TCompactProtocol;
    import org.apache.thrift.protocol.TMultiplexedProtocol;
    import org.apache.thrift.protocol.TProtocol;
    import org.apache.thrift.transport.TFramedTransport;
    import org.apache.thrift.transport.TSocket;
    import org.apache.thrift.transport.TTransport;
    
    import com.test.Calculator;
    import com.test.Operation;
    import com.test.Sender;
    import com.test.Work;
    
    public class ClientMain {
    	public static void main(String[] args) {
    		doClient();
    	}
    
    	public static void performs(Calculator.Client client) throws TException {
    		Work work1 = new Work();
    
    		work1.op = Operation.DIVIDE;
    		work1.num1 = 1;
    		work1.num2 = 0;
    
    		Work work2 = new Work();
    
    		work2.op = Operation.SUBTRACT;
    		work2.num1 = 15;
    		work2.num2 = 10;
    
    		List<Work> workList = new ArrayList<Work>();
    		workList.add(work1);
    		workList.add(work2);
    
    		int r1 = client.addList(workList);
    		System.out.println("addList=" + r1);
    
    		int retAdd = client.add(1, 2);
    		System.out.println("retAdd=" + retAdd);
    	}
    
    	public static void performs(Sender.Client client) throws TException {
    		client.ping();
    		System.out.println("ping ok");
    	}
    
    	public static void doClient() {
    		try {
    			TTransport transport = new TFramedTransport(new TSocket("localhost", 9090));
    
    			TProtocol protocol1 = new TMultiplexedProtocol(new TCompactProtocol(transport), "Calculator");
    			Calculator.Client client1 = new Calculator.Client(protocol1);
    
    			TProtocol protocol2 = new TMultiplexedProtocol(new TCompactProtocol(transport), "Sender");
    			Sender.Client client2 = new Sender.Client(protocol2);
    
    			transport.open();
    
    			performs(client1);
    			performs(client2);
    
    			transport.close();
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    	}
    }