基於Netty實現服務端與客戶端通信

本文轉載自查看原文 2020-04-13 03:01 1578 服務端/ Netty/ 客戶端/ Android

個人博客

http://www.milovetingting.cn

基於Netty實現服務端與客戶端通信

前言

本文介紹基於Netty實現的服務端與客戶端通信的簡單使用方法，並在此基礎上實現一個簡單的服務端-客戶端指令通信的Demo。

Netty是什么

Netty是一個NIO客戶端-服務器框架，可以快速輕松地開發網絡應用程序，例如協議服務器和客戶端。它極大地簡化了網絡編程，例如TCP和UDP套接字服務器的開發。提供一個異步事件驅動的網絡應用程序框架和工具，以快速開發可維護的高性能和高可擴展性協議服務器和客戶端。

以上內容摘選自https://netty.io/wiki/user-guide-for-4.x.html

Netty具有以下特點：

適用於各種傳輸類型的統一API-阻塞和非阻塞套接字
更高的吞吐量，更低的延遲
減少資源消耗
減少不必要的內存復制
完整的SSL / TLS和StartTLS支持

以上內容摘選自https://netty.io/

使用入門

Netty的使用，可以參照Netty的官方文檔，這里以4.x為例來演示Netty在服務端和客戶端上使用。文檔地址：https://netty.io/wiki/user-guide-for-4.x.html

這里用Eclipse來進行開發,服務端和客戶端都放在一個工程里。

新建Java工程

服務端

首先需要導入netty的jar包。這里使用netty-all-4.1.48.Final.jar。

NettyServer

新建NettyServer類

public class NettyServer {

	private int mPort;

	public NettyServer(int port) {
		this.mPort = port;
	}

	public void run() {
		EventLoopGroup bossGroup = new NioEventLoopGroup();
		EventLoopGroup workerGroup = new NioEventLoopGroup();
		try {
			ServerBootstrap b = new ServerBootstrap();
			b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
					// 指定連接隊列大小
					.option(ChannelOption.SO_BACKLOG, 128)
					//KeepAlive
					.childOption(ChannelOption.SO_KEEPALIVE, true)
					//Handler
					.childHandler(new ChannelInitializer<SocketChannel>() {

						@Override
						protected void initChannel(SocketChannel channel) throws Exception {
							channel.pipeline().addLast(new NettyServerHandler());
						}
					});
			ChannelFuture f = b.bind(mPort).sync();
			if (f.isSuccess()) {
				LogUtil.log("Server,啟動Netty服務端成功，端口號:" + mPort);
			}
			// f.channel().closeFuture().sync();
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			// workerGroup.shutdownGracefully();
			// bossGroup.shutdownGracefully();
		}
	}

}

NettyServerHandler

在初始化時，需要指定Handle，用來處理Channel相關業務。

public class NettyServerHandler extends ChannelInboundHandlerAdapter {

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		LogUtil.log("Server,channelActive");
	}

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		LogUtil.log("Server,接收到客戶端發來的消息:" + msg);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		LogUtil.log("Server,exceptionCaught");
		cause.printStackTrace();
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		LogUtil.log("Server,channelInactive");
	}

}

經過上面這些步驟后，服務端最基本的設置就完成了。

客戶端

客戶端和服務端在初始化時大體是類似的，不過相比服務端要簡單一些。

NettyClient

public class NettyClient {

	private String mHost;

	private int mPort;

	private NettyClientHandler mClientHandler;

	private ChannelFuture mChannelFuture;

	public NettyClient(String host, int port) {
		this.mHost = host;
		this.mPort = port;
	}

	public void connect() {
		EventLoopGroup workerGroup = new NioEventLoopGroup();
		try {
			Bootstrap b = new Bootstrap();
			mClientHandler = new NettyClientHandler();
			b.group(workerGroup).channel(NioSocketChannel.class)
					// KeepAlive
					.option(ChannelOption.SO_KEEPALIVE, true)
					// Handler
					.handler(new ChannelInitializer<SocketChannel>() {

						@Override
						protected void initChannel(SocketChannel channel) throws Exception {
							channel.pipeline().addLast(mClientHandler);
						}
					});
			mChannelFuture = b.connect(mHost, mPort).sync();
			if (mChannelFuture.isSuccess()) {
				LogUtil.log("Client,連接服務端成功");
			}
			mChannelFuture.channel().closeFuture().sync();
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			workerGroup.shutdownGracefully();
		}
	}
}

NettyClientHandler

public class NettyClientHandler extends ChannelInboundHandlerAdapter {

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		LogUtil.log("Client,channelActive");
	}

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		LogUtil.log("Client,接收到服務端發來的消息:" + msg);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		LogUtil.log("Client,exceptionCaught");
		cause.printStackTrace();
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		LogUtil.log("Client,channelInactive");
	}

}

到這里，客戶端最基本設置就完成了。

連接服務端

新建一個Main類，用於測試服務端和客戶端是否能正常連接。

public class Main {

	public static void main(String[] args) {
		try {
			String host = "127.0.0.1";
			int port = 12345;
			NettyServer server = new NettyServer(port);
			server.run();
			Thread.sleep(1000);
			NettyClient client = new NettyClient(host, port);
			client.connect();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}

運行main方法，輸出日志如下：

2020-4-13 0:11:02--Server,啟動Netty服務端成功，端口號:12345
2020-4-13 0:11:03--Client,channelActive
2020-4-13 0:11:03--Client,連接服務端成功
2020-4-13 0:11:03--Server,channelActive

可以看到，客戶端成功連接上了服務端，服務端和客戶端里設置的Handler的channelActive方法都會回調。

服務端與客戶端通信

在服務端與客戶端連接成功后，我們往往需要在雙方間進行通信。這里假定，在連接成功后，服務端給客戶端發送一個歡迎信息"你好，客戶端"，而客戶端在收到服務端的消息后，也給服務端回復一個消息"你好，服務端"。下面來實現具體的功能。

修改服務端NettyServerHandler中的channelActive方法和channelRead方法,在channelActive方法中給客戶端發送消息，在channelRead方法中解析客戶端發來的消息

public class NettyServerHandler extends ChannelInboundHandlerAdapter {

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		LogUtil.log("Server,channelActive");
		ByteBuf byteBuf = Unpooled.copiedBuffer("你好，客戶端", Charset.forName("utf-8"));
		ctx.writeAndFlush(byteBuf);
	}

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		ByteBuf buf = (ByteBuf) msg;
		byte[] buffer = new byte[buf.readableBytes()];
		buf.readBytes(buffer);
		String message = new String(buffer, "utf-8");
		LogUtil.log("Server,接收到客戶端發來的消息:" + message);
	}

}

修改客戶端NettyClientHandler中的channelRead方法，當收到服務端的消息時，回復服務端

public class NettyClientHandler extends ChannelInboundHandlerAdapter {

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		ByteBuf buf = (ByteBuf) msg;
		byte[] buffer = new byte[buf.readableBytes()];
		buf.readBytes(buffer);
		String message = new String(buffer,"utf-8");
		LogUtil.log("Client,接收到服務端發來的消息:" + message);
		
		ByteBuf byteBuf = Unpooled.copiedBuffer("你好，服務端", Charset.forName("utf-8"));
		ctx.writeAndFlush(byteBuf);
	}

}

運行后，輸出日志如下：

2020-4-13 0:29:16--Server,啟動Netty服務端成功，端口號:12345
2020-4-13 0:29:17--Client,channelActive
2020-4-13 0:29:17--Client,連接服務端成功
2020-4-13 0:29:17--Server,channelActive
2020-4-13 0:29:17--Client,接收到服務端發來的消息:你好，客戶端
2020-4-13 0:29:17--Server,接收到客戶端發來的消息:你好，服務端

可以看到，服務端與客戶端已經可以正常通信。

粘包與拆包

在實際的使用場景中，可能會存在短時間內大量數據發送的問題。我們模擬這個場景。在客戶端連接上服務端后，服務端給客戶端發送100個消息，而為便於分析，客戶端在收到服務端消息后，不作回復。

修改服務端中NettyServerHandler的channelActive方法

@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		LogUtil.log("Server,channelActive");
		for (int i = 0; i < 100; i++) {
			ByteBuf byteBuf = Unpooled.copiedBuffer("你好，客戶端", Charset.forName("utf-8"));
			ctx.writeAndFlush(byteBuf);
		}
	}

修改客戶端中NettyClientHandler的channelRead方法

@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		ByteBuf buf = (ByteBuf) msg;
		byte[] buffer = new byte[buf.readableBytes()];
		buf.readBytes(buffer);
		String message = new String(buffer, "utf-8");
		LogUtil.log("Client,接收到服務端發來的消息:" + message);

        //ByteBuf byteBuf = Unpooled.copiedBuffer("你好，服務端", Charset.forName("utf-8"));
        //ctx.writeAndFlush(byteBuf);
	}

運行后，輸出的部分結果如下:

2020-4-13 0:35:28--Server,啟動Netty服務端成功，端口號:12345
2020-4-13 0:35:29--Client,channelActive
2020-4-13 0:35:29--Client,連接服務端成功
2020-4-13 0:35:29--Server,channelActive
2020-4-13 0:35:29--Client,接收到服務端發來的消息:你好，客戶端
2020-4-13 0:35:29--Client,接收到服務端發來的消息:你好，客戶端你好，客戶端你好，客戶端你好，客戶端你好，客戶端你好，客戶端你好，客戶端你好，客戶端你好，客戶端你好，客戶端你好，客戶端你好，客戶端你好，客戶端你好，客戶端你好，客戶端你好，客戶端你好，客戶端
2020-4-13 0:35:29--Client,接收到服務端發來的消息:你好，客戶端

可以看到，出現了多條消息"粘"在一起的情況。

什么是粘包與拆包

TCP是個"流"協議，所謂流，就是沒有界限的一串數據。TCP底層並不了解上層業務數據的具體含義，它會根據TCP緩沖區的實際情況進行包的划分，所以在業務上認為，一個完整的包可能會被TCP拆分成多個包進行發送，也有可能把多個小的包封裝成一個大的數據包發送，這就是所謂的TCP粘包和拆包問題。

以上內容摘選自TCP粘包/拆包與Netty解決方案

解決方案

在沒有 Netty 的情況下，用戶如果自己需要拆包，基本原理就是不斷從 TCP 緩沖區中讀取數據，每次讀取完都需要判斷是否是一個完整的數據包如果當前讀取的數據不足以拼接成一個完整的業務數據包，那就保留該數據，繼續從 TCP 緩沖區中讀取，直到得到一個完整的數據包。如果當前讀到的數據加上已經讀取的數據足夠拼接成一個數據包，那就將已經讀取的數據拼接上本次讀取的數據，構成一個完整的業務數據包傳遞到業務邏輯，多余的數據仍然保留，以便和下次讀到的數據嘗試拼接。

以上內容摘選自徹底理解Netty，這一篇文章就夠了

而使用Netty，則解決這個問題的方法就簡單多了。Netty已經提供了四個拆包器:

FixedLengthFrameDecoder:固定長度的拆包器,Netty會把固定長度的數據包發送給下一個channelHandler
LineBasedFrameDecoder:行拆包器,每個數據包以換行符分隔發送
DelimiterBasedFrameDecoder:分隔符拆包器，可以自定義分隔符，行拆包器是分隔符拆包器的一種特例
LengthFieldBasedFrameDecoder：基於長度域的拆包器，如果自定義協議中包含長度域的字段，就可以使用這個拆包器

在這里，我們選用分隔符拆包器

首先定義分隔符

public class Config {
	public static final String DATA_PACK_SEPARATOR = "#$&*";
}

在服務端的channelHandler配置中，需要增加

@Override
protected void initChannel(SocketChannel channel) throws Exception {
    //這個配置需要在添加Handler前設置
	channel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,Unpooled.copiedBuffer(Config.DATA_PACK_SEPARATOR.getBytes())));
	channel.pipeline().addLast(new NettyServerHandler());
	}

在客戶端的channelHandler的配置中，同樣也需要增加

@Override
protected void initChannel(SocketChannel channel) throws Exception {
    //這個配置需要在添加Handler前設置
	channel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,Unpooled.copiedBuffer(Config.DATA_PACK_SEPARATOR.getBytes())));
	channel.pipeline().addLast(new NettyServerHandler());
	}

發送數據時，在數據的末尾增加分隔符:

@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		LogUtil.log("Server,channelActive");
		for (int i = 0; i < 100; i++) {
			ByteBuf byteBuf = Unpooled.copiedBuffer("你好，客戶端"+Config.DATA_PACK_SEPARATOR, Charset.forName("utf-8"));
			ctx.writeAndFlush(byteBuf);
		}
	}

運行后，可以發現，已經解決"粘包"與"拆包"的問題。

心跳

在網絡應用中，為了判斷連接是否還存在，一般會通過發送心跳包來檢測。在Netty中，配置心跳包的步驟如下

在客戶端的channelHandler的配置中，需要增加

@Override
protected void initChannel(SocketChannel channel) throws Exception {
			channel.pipeline().addLast(new IdleStateHandler(5, 5, 10));
            //...
						}

在NettyClientHandler中，重寫userEventTriggered方法

@Override
	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
		IdleStateEvent event = (IdleStateEvent) evt;
		LogUtil.log("Client,Idle:" + event.state());
		switch (event.state()) {
		case READER_IDLE:

			break;
		case WRITER_IDLE:
			ByteBuf byteBuf = Unpooled.copiedBuffer("心跳^v^v", Charset.forName("utf-8"));
			break;
		case ALL_IDLE:
			break;
		default:
			super.userEventTriggered(ctx, evt);
			break;
		}
	}

當寫空閑達到配置的時間時，往服務端發送一個心跳消息

運行后，日志輸出如下:

2020-4-13 1:22:50--Server,啟動Netty服務端成功，端口號:12345
2020-4-13 1:22:51--Client,channelActive
2020-4-13 1:22:51--Client,連接服務端成功
2020-4-13 1:22:51--Server,channelActive
2020-4-13 1:22:51--Client,接收到服務端發來的消息:你好，客戶端
2020-4-13 1:22:56--Client,Idle:WRITER_IDLE
2020-4-13 1:22:56--Server,接收到客戶端發來的消息:心跳^v^
2020-4-13 1:22:56--Client,Idle:READER_IDLE
2020-4-13 1:23:01--Client,Idle:WRITER_IDLE
2020-4-13 1:23:01--Server,接收到客戶端發來的消息:心跳^v^
2020-4-13 1:23:01--Client,Idle:READER_IDLE

可以看到，心跳包按我們配置的時間正常輸出了。

配置編碼器與解碼器

我們上面在發送數據時，需要通過ByteBuf來轉換String，而通過配置編碼，解碼器，我們就可以直接發送字符串。配置如下：

在服務端與客戶端的channelHandler分別增加以下配置：

@Override
protected void initChannel(SocketChannel channel) throws Exception {
	//...
	//這個配置需要在添加Handler前設置
	channel.pipeline().addLast("encoder", new StringEncoder());
	channel.pipeline().addLast("decoder", new StringDecoder());
    //...
}

在發送消息時，則可以直接通過ctx.writeAndFlush("心跳^v^" + Config.DATA_PACK_SEPARATOR)的形式來發送。

源碼

到此，最簡單的服務端與客戶端通信的Demo已經完成。源碼地址：https://github.com/milovetingting/Samples/tree/master/NettyDemo

使用進階

在上面的基礎上，我們來實現一個下面的需求：

客戶端需要登錄到服務端
客戶端登錄成功后，服務端可以給客戶端發送指令消息，客戶端在收到消息及處理完消息后，都需要上報給服務端

封裝連接

為便於程序擴展，我們將客戶端連接服務端的部分抽取出來。通過一個接口來定義連接的方法，而連接的具體實現由子類來實現。

定義接口

public interface IConnection {

	/**
	 * 連接服務器
	 * 
	 * @param host     服務器地址
	 * @param port     端口
	 * @param callback 連接回調
	 */
	public void connect(String host, int port, IConnectionCallback callback);

}

在這里還需要定義連接的回調接口

public interface IConnectionCallback {

	/**
	 * 連接成功
	 */
	public void onConnected();

}

具體的連接實現類

public class NettyConnection implements IConnection {

	private NettyClient mClient;

	@Override
	public void connect(String host, int port, IConnectionCallback callback) {
		if (mClient == null) {
			mClient = new NettyClient(host, port);
			mClient.setConnectionCallBack(callback);
			mClient.connect();
		}
	}

}

為便於管理連接，定義一個連接的管理類

public class ConnectionManager implements IConnection {

	private static IConnection mConnection;

	private ConnectionManager() {

	}

	static class ConnectionManagerInner {
		private static ConnectionManager INSTANCE = new ConnectionManager();
	}

	public static ConnectionManager getInstance() {
		return ConnectionManagerInner.INSTANCE;
	}

	public static void initConnection(IConnection connection) {
		mConnection = connection;
	}

	private void checkInit() {
		if (mConnection == null) {
			throw new IllegalAccessError("please invoke initConnection first!");
		}
	}

	@Override
	public void connect(String host, int port, IConnectionCallback callback) {
		checkInit();
		mConnection.connect(host, port, callback);
	}

}

調用連接：

public class Main {

	public static void main(String[] args) {
		try {
			String host = "127.0.0.1";
			int port = 12345;
			NettyServer server = new NettyServer(port);
			server.run();
			Thread.sleep(1000);
			ConnectionManager.initConnection(new NettyConnection());
			ConnectionManager.getInstance().connect(host, port, new IConnectionCallback() {

				@Override
				public void onConnected() {
					LogUtil.log("Main,onConnected"););
				}
			});
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

}

在調用connect方法前，需要先調用initConnection來指定具體的連接類

消息Bean的定義

在連接成功后，服務端會給客戶端發送一個歡迎的消息。為便於管理，我們定義一個消息Bean

public class Msg {

	/**
	 * 歡迎
	 */
	public static final int TYPE_WELCOME = 0;

	public int type;

	public String msg;

}

服務端發送歡迎消息

服務端發送消息

public class NettyServerHandler extends ChannelInboundHandlerAdapter {

	private ChannelHandlerContextWrapper mChannelHandlerContextWrapper;

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		LogUtil.log("Server,channelActive");
		mChannelHandlerContextWrapper = new ChannelHandlerContextWrapper(ctx);
		MsgUtil.sendWelcomeMsg(mChannelHandlerContextWrapper);
	}
}

在這里，通過定義一個ChannelHandlerContextWrapper類來統一管理消息分隔符

public class ChannelHandlerContextWrapper {

	private ChannelHandlerContext mContext;

	public ChannelHandlerContextWrapper(ChannelHandlerContext context) {
		this.mContext = context;
	}

	/**
	 * 包裝writeAndFlush方法
	 * 
	 * @param object
	 */
	public void writeAndFlush(Object object) {
		mContext.writeAndFlush(object + Config.DATA_PACK_SEPARATOR);
	}

}

再進一步，通過定義MsgUtil類來封裝發送歡迎消息

public class MsgUtil {

	/**
	 * 發送歡迎消息
	 * 
	 * @param wrapper
	 */
	public static void sendWelcomeMsg(ChannelHandlerContextWrapper wrapper) {
		Msg msg = new Msg();
		msg.type = Msg.TYPE_WELCOME;
		msg.msg = "你好，客戶端";
		wrapper.writeAndFlush(Global.sGson.toJson(msg));
	}

}

客戶端消息接收

對於客戶端而言，為方便處理消息，我們需要定義一個方法來接收消息。通過在IConnection接口中新增一個registerMsgCallback方法來實現

public interface IConnection {

	/**
	 * 連接服務器
	 * 
	 * @param host     服務器地址
	 * @param port     端口
	 * @param callback 連接回調
	 */
	public void connect(String host, int port, IConnectionCallback callback);

	/**
	 * 注冊消息回調
	 * 
	 * @param callback
	 */
	public void registerMsgCallback(IMsgCallback callback);

}

在這里，還需要新增IMsgCallback接口

public interface IMsgCallback {

	/**
	 * 接收到消息時的回調
	 * 
	 * @param msg
	 */
	public void onMsgReceived(Msg msg);

}

對應到實現類

public class NettyConnection implements IConnection {

	private NettyClient mClient;

	@Override
	public void connect(String host, int port, IConnectionCallback callback) {
		if (mClient == null) {
			mClient = new NettyClient(host, port);
			mClient.setConnectionCallBack(callback);
			mClient.connect();
		}
	}

	@Override
	public void registerMsgCallback(IMsgCallback callback) {
		if (mClient == null) {
			throw new IllegalAccessError("please invoke connect first!");
		}
		mClient.registerMsgCallback(callback);
	}

}

消息的分發

在客戶端，為便於處理消息，我們對消息類型進行划分

修改消息Bean

public class Msg {

	/**
	 * 歡迎
	 */
	public static final int TYPE_WELCOME = 0;

	/**
	 * 心跳
	 */
	public static final int TYPE_HEART_BEAT = 1;

	/**
	 * 登錄
	 */
	public static final int TYPE_LOGIN = 2;

	public static final int TYPE_COMMAND_A = 3;

	public static final int TYPE_COMMAND_B = 4;

	public static final int TYPE_COMMAND_C = 5;

	public int type;

	public String msg;
}

假定消息是串行的，需要一個一個地處理。為便於管理消息，增加MsgQueue類

public class MsgQueue {

	private PriorityBlockingQueue<Msg> mQueue;

	private boolean using;

	private MsgQueue() {
		mQueue = new PriorityBlockingQueue<>(128, new Comparator<Msg>() {

			@Override
			public int compare(Msg msg1, Msg msg2) {
				int res = msg2.priority - msg1.priority;
				if (res == 0 && msg1.time != msg2.time) {
					return (int) (msg2.time - msg1.time);
				}
				return res;
			}
		});
	}

	public static MsgQueue getInstance() {
		return MsgQueueInner.INSTANCE;
	}

	private static class MsgQueueInner {
		private static final MsgQueue INSTANCE = new MsgQueue();
	}

	/**
	 * 將消息加入消息隊列
	 * 
	 * @param msg
	 */
	public void enqueueMsg(Msg msg) {
		mQueue.add(msg);
	}

	/**
	 * 從消息隊列獲取消息
	 * 
	 * @return
	 */
	public synchronized Msg next() {
		if (using) {
			return null;
		}
		Msg msg = mQueue.poll();
		if (msg != null) {
			makeUse(true);
		}
		return msg;
	}

	/**
	 * 標記使用狀態
	 * 
	 * @param use
	 */
	public synchronized void makeUse(boolean use) {
		using = use;
	}

	/**
	 * 是否能夠使用
	 * 
	 * @return
	 */
	public synchronized boolean canUse() {
		return !using;
	}

}

增加消息的分發類MsgDispatcher

public class MsgDispatcher {

	private static Map<Integer, Class<? extends IMsgHandler>> mHandlerMap;

	static {
		mHandlerMap = new HashMap<>();
		mHandlerMap.put(Msg.TYPE_WELCOME, WelcomeMsgHandler.class);
		mHandlerMap.put(Msg.TYPE_HEART_BEAT, HeartBeatMsgHandler.class);
		mHandlerMap.put(Msg.TYPE_LOGIN, HeartBeatMsgHandler.class);
		mHandlerMap.put(Msg.TYPE_COMMAND_A, CommandAMsgHandler.class);
		mHandlerMap.put(Msg.TYPE_COMMAND_B, CommandBMsgHandler.class);
		mHandlerMap.put(Msg.TYPE_COMMAND_C, CommandCMsgHandler.class);
	}

	public static void dispatch() {
		if (MsgQueue.getInstance().canUse()) {
			Msg msg = MsgQueue.getInstance().next();
			if (msg == null) {
				return;
			}
			dispatch(msg);
		}
	}

	public static void dispatch(Msg msg) {
		try {
			IMsgHandler handler = (IMsgHandler) Class.forName(mHandlerMap.get(msg.type).getName()).newInstance();
			handler.handle(msg);
		} catch (InstantiationException e) {
			e.printStackTrace();
		} catch (IllegalAccessException e) {
			e.printStackTrace();
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		}
	}

}

消息的處理

定義IMsgHandler,在這里定義了處理的方法，具體實現由子類實現

public interface IMsgHandler {

	/**
	 * 處理消息
	 * 
	 * @param msg
	 */
	public void handle(Msg msg);

}

為統一管理，定義Base類BaseCommandHandler

public abstract class BaseCommandHandler implements IMsgHandler {

	@Override
	public void handle(Msg msg) {
		execute(msg);
	}

	public final void execute(Msg msg) {
		LogUtil.log("Client,received command:" + msg);
		doHandle(msg);
		MsgQueue.getInstance().makeUse(false);
		LogUtil.log("Client,report command:" + msg);
		MsgDispatcher.dispatch();
	}

	public abstract void doHandle(Msg msg);

}

在BaseCommandHandler中，定義execute方法，順序調用：上報消息已接收成功、處理消息、上報消息已處理完成。這里的消息上報部分，都只是輸出一個日志來代替，在實際的業務中，可以抽取出一個抽象方法，讓子類來實現。

定義子類，繼承自BaseCommandHandler

public class LoginMsgHandler extends BaseCommandHandler {

	@Override
	public void doHandle(Msg msg) {
		LogUtil.log("Client,handle msg:" + msg);
	}

}

對應的心跳類型消息、歡迎類型消息等，都可以新增對應的處理類來實現，這里不再展開。

接收到消息時的處理

public class Main {

	public static void main(String[] args) {
		try {
			String host = "127.0.0.1";
			int port = 12345;
			NettyServer server = new NettyServer(port);
			server.run();
			Thread.sleep(1000);
			ConnectionManager.initConnection(new NettyConnection());
			ConnectionManager.getInstance().connect(host, port, new IConnectionCallback() {

				@Override
				public void onConnected() {
					LogUtil.log("Main,onConnected");

					ConnectionManager.getInstance().registerMsgCallback(new IMsgCallback() {

						@Override
						public void onMsgReceived(Msg msg) {
							MsgQueue.getInstance().enqueueMsg(msg);
							MsgDispatcher.dispatch();
						}
					});
				}
			});
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

}

客戶端登錄

修改消息Bean，增加登錄的請求和響應

public class Msg {

	/**
	 * 歡迎
	 */
	public static final int TYPE_WELCOME = 0;

	/**
	 * 心跳
	 */
	public static final int TYPE_HEART_BEAT = 1;

	/**
	 * 登錄
	 */
	public static final int TYPE_LOGIN = 2;

	public static final int TYPE_COMMAND_A = 3;

	public static final int TYPE_COMMAND_B = 4;

	public static final int TYPE_COMMAND_C = 5;

	public int type;

	public String msg;

	public int priority;

	public long time;

	/**
	 * 登錄請求信息
	 * 
	 * @author Administrator
	 *
	 */
	public static class LoginRuquestInfo {
		/**
		 * 用戶名
		 */
		public String user;

		/**
		 * 密碼
		 */
		public String pwd;

		@Override
		public String toString() {
			return "LoginRuquestInfo [user=" + user + ", pwd=" + pwd + "]";
		}
	}

	/**
	 * 登錄響應信息
	 * 
	 * @author Administrator
	 *
	 */
	public static class LoginResponseInfo {

		/**
		 * 登錄成功
		 */
		public static final int CODE_SUCCESS = 0;

		/**
		 * 登錄失敗
		 */
		public static final int CODE_FAILED = 100;

		/**
		 * 響應碼
		 */
		public int code;

		/**
		 * 響應數據
		 */
		public String data;

		public static class ResponseData {
			public String token;
		}

		@Override
		public String toString() {
			return "LoginResponseInfo [code=" + code + ", data=" + data + "]";
		}

	}
}

發送登錄請求

public class Main {

	public static void main(String[] args) {
		try {
			String host = "127.0.0.1";
			int port = 12345;
			NettyServer server = new NettyServer(port);
			server.run();
			Thread.sleep(1000);
			ConnectionManager.initConnection(new NettyConnection());
			ConnectionManager.getInstance().connect(host, port, new IConnectionCallback() {

				@Override
				public void onConnected() {
					LogUtil.log("Main,onConnected");

					ConnectionManager.getInstance().registerMsgCallback(new IMsgCallback() {

						@Override
						public void onMsgReceived(Msg msg) {
							MsgQueue.getInstance().enqueueMsg(msg);
							MsgDispatcher.dispatch();
						}
					});

					Msg msg = new Msg();
					msg.type = Msg.TYPE_LOGIN;

					Msg.LoginRuquestInfo request = new LoginRuquestInfo();
					request.user = "wangyz";
					request.pwd = "wangyz";

					Gson gson = new Gson();
					msg.msg = gson.toJson(request);

					ConnectionManager.getInstance().sendMsg(msg);
				}
			});
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

}

這里，引入Gson，將消息Bean轉成json字符串后發送。

對應到服務端，為便於解析出消息，也需要對應的修改消息的Bean。服務端對消息的具體分發與處理，和客戶端類似，這里不再展開。

源碼

由於篇幅限制，Demo中指令的優先級處理，模擬服務端指令下發等，這里沒有再進一步詳細介紹，具體可以參考源碼:https://github.com/milovetingting/Samples/tree/master/Netty

后記

本文介紹了基於Netty實現服務端與客戶端通信的基本用法，以及在此基礎上，實現處理服務端指令並上報。Demo中通信的數據格式，用到了json，而優化的做法，可以用protobuf來實現，這里只展示通信的流程及簡單的封裝，因而未使用protobuf。Demo中只實現大體的流程，可能存在未測試到的Bug，權當一個參考的思路吧。

End~

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 Netty入門——客戶端與服務端通信 Netty實現客戶端和服務端通信簡單例子 netty實現客戶端服務端心跳重連使用WebSocket實現服務端和客戶端的通信使用WebSocket實現服務端和客戶端的通信 python實現一個客戶端與服務端的通信 netty4----netty5的客戶端和服務端 Netty中使用WebSocket實現服務端與客戶端的長連接通信發送消息 netty學習第5章 netty整合websocket實現服務端與客戶端消息推送 socket通信之客戶端與服務端通信