ToB企服应用市场:ToB评测及商务社交产业平台

标题: Netty实现udp服务器 [打印本页]

---

作者: 科技颠覆者    时间: 2024-6-28 23:01
标题: Netty实现udp服务器
1、TCP与UDP通讯协议

网络传输层协议有两种，一种是TCP，别的一种是UDP。
TCP是一种面向连接的协议，提供可靠的数据传输。TCP通过三次握手建立连接，并通过确认和重传机制，包管数据的完备性和可靠性。TCP适用于对数据正确性要求较高、对实时性要求较低的应用场景，如网页浏览、文件传输等。
UDP是一种无连接的协议，不包管数据的可靠性传输。UDP通过尽力交付数据包的方式进行传输，不对数据包的传输状态进行确认和重传，因此速率较快。UDP适用于对实时性要求较高、对数据正确性要求较低的应用场景，如视频传输、语音通讯等。
那么，这两种有哪些区别呢？请看下面：

• 连接方式：UDP是无连接的，TCP是面向连接的；
  
• 可靠性：UDP不包管数据的可靠性传输，TCP包管数据的可靠传输；
  
• 速率：UDP传输速率较快，TCP传输速率较慢；
  
• 传输方式：UDP采用尽力交付的方式传输数据包，不进行确认和重传；TCP通过确认和重传机制包管数据的完备性和可靠性；
  
• 开销：UDP的开销较低，TCP的开销较高。
  

总结：我们可以根据具体的应用场景和需求选择使用UDP或TCP进行数据传输。如果对数据的实时性要求较高，且对数据正确性要求较低，可以选择使用UDP。如果对数据的正确性要求较高，可以选择使用TCP。
2、游戏行业选择的通讯协议

游戏由于对数据的正确性（不允许丢包，乱序）非常高，一样平常都是选择基于TCP的socket通讯。但UDP的低时延，快速传输对实时性要求非常高的游戏类型也黑白常大的吸引力。据说，魔兽天下以及Dota2使用UDP开发。当然，使用udp通讯的游戏肯定在通讯层做了适配，包管关键数据不丢包，不乱序。
比年来，基于UDP协议的KCP（Kuai Control Protocol，快速可靠传输协议），也得到了快速的发展。据说，原神是使用KCP通讯的。
3、Netty使用UDP协议

netty使用udp协议，网上的例子都黑白常简朴的。都是两个类搞定。没有办理以下几个题目：

• 如何与现有网络框架适配（支持私有协议栈，支持javabean，不单单是字符串）
  
• 如何自动与客户端通讯（不再是客户端发一个消息，服务器直接推送一个回包）
  

本文重要就这两个题目进行案例说明。
3.1、netty数据包载体

udp是无连接的，这意味着通讯双方无需像TCP那般“三次握手四次释放”。只要知道对方的socket地址（Ip+Port），即可发送数据，发完即终止。在Netty里，udp的通讯载体叫做DatagramPacket，负责将数据（ByteBuf）从源头发送到目的地。

1. public class DatagramPacket extends DefaultAddressedEnvelope<ByteBuf, InetSocketAddress> implements ByteBufHolder {
2.     public DatagramPacket(ByteBuf data, InetSocketAddress recipient) {
3.         super(data, recipient);
4.     }
6.     public DatagramPacket(ByteBuf data, InetSocketAddress recipient, InetSocketAddress sender) {
7.         super(data, recipient, sender);
8.     }
9. }

复制代码

3.2私有协议栈编解码

3.2.1、通讯基类

而我们的网络底层通讯是基于javabean的，因此定义我们的消息基类如下：

1. public class UdpMessage implements Message {
3.     private String senderIp;
5.     private int senderPort;
7.     private String receiverIp;
9.     private int receiverPort;
11. }

复制代码

3.2.2、私有协议栈消息编码

我们将私有协议栈定义为 包头（消息类型id），包体（具体消息经编码后的字节数组）。将自定义消息UdpMessage转为DatagramPacket。

1. public class UdpProtocolEncoder extends MessageToMessageEncoder<UdpMessage> {
2.     private static final Logger logger = LoggerFactory.getLogger("socketserver");
4.     private final MessageFactory messageFactory;
6.     private final MessageCodec messageCodec;
8.     public UdpProtocolEncoder(MessageFactory messageFactory, MessageCodec messageCodec) {
9.         this.messageFactory = messageFactory;
10.         this.messageCodec = messageCodec;
11.     }
13.     @Override
14.     protected void encode(ChannelHandlerContext ctx, UdpMessage message, List<Object> out) throws Exception {
15.         // ----------------protocol pattern-------------------------
16.         // packetLength | cmd | body
17.         // int int byte[]
18.         int  cmd = messageFactory.getMessageId(message.getClass());
19.         try {
20.             byte[] body = messageCodec.encode(message);
21.             //消息内容长度
22.             ByteBuf buf = Unpooled.buffer(body.length+4);
23.             // 写入cmd类型
24.             buf.writeInt(cmd);
25.             buf.writeBytes(body);
26.             out.add(new DatagramPacket(buf, new InetSocketAddress(message.getReceiverIp(), message.getReceiverPort())));
27.         } catch (Exception e) {
28.             logger.error("wrote message {} failed", cmd, e);
29.         }
30.     }
31. }

复制代码

这里消息pojo编码，使用了jforgame的组件，根据javabean的字段元信息，自动编码为byte数组。
依靠说明如下：

1. <dependency>
2.     <groupId>io.github.jforgame</groupId>
3.     <artifactId>jforgame-codec-struct</artifactId>
4.     <version>1.1.0</version>
5. </dependency>

复制代码

3.2.3、私有协议栈消息解码

私有协议栈解码负责将数据包DatagramPacket转为UdpMessage。将底层数据流转为ByteBuf之后，还必要将字节数据进行解码，才可以转换为应用程序熟悉的消息。

1. public class UdpProtocolDecoder extends MessageToMessageDecoder<DatagramPacket> {
2.     private final MessageFactory messageFactory;
4.     private final MessageCodec messageCodec;
7.     public UdpProtocolDecoder(MessageFactory messageFactory, MessageCodec messageCodec) {
8.         this.messageFactory = messageFactory;
9.         this.messageCodec = messageCodec;
10.     }
12.     @Override
13.     protected void decode(ChannelHandlerContext ctx, DatagramPacket msg, List<Object> out) throws Exception {
14.         ByteBuf in = msg.content();
15.         int length = in.readableBytes();
16.         int cmd = in.readInt();
17.         byte[] body = new byte[length - 4];
18.         in.readBytes(body);
19.         Class<?> msgClazz = messageFactory.getMessage(cmd);
20.         out.add(messageCodec.decode(msgClazz, body));
21.     }
23. }

复制代码

4、服务端代码

4.1、会话管理

服务端会话管理（建立及摧毁），以及消息担当（下文的channelRead方法）。

1. @ChannelHandler.Sharable
2. public class UdpChannelIoHandler extends ChannelInboundHandlerAdapter {
4.     private final static Logger logger = LoggerFactory.getLogger("socketserver");
6.     /** 消息分发器 */
7.     private final SocketIoDispatcher messageDispatcher;
9.     public UdpChannelIoHandler(SocketIoDispatcher messageDispatcher) {
10.         super();
11.         this.messageDispatcher = messageDispatcher;
12.     }
14.     @Override
15.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
16.         Channel channel = ctx.channel();
17.         ChannelUtils.duplicateBindingSession(ctx.channel(), new NSession(channel));
18.         SessionManager.getInstance().buildSession(ChannelUtils.getSessionBy(channel));
19.         System.out.println("socket register " + channel);
20.     }
22.     @Override
23.     public void channelRead(ChannelHandlerContext context, Object packet) throws Exception {
24.         logger.debug("receive pact, content is {}", packet.getClass().getSimpleName());
26.         final Channel channel = context.channel();
27.         IdSession session = ChannelUtils.getSessionBy(channel);
28.         messageDispatcher.dispatch(session, packet);
29.     }
31.     @Override
32.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
33.         Channel channel = ctx.channel();
34.         System.out.println("socket inactive " + channel);
35.         IdSession userSession = ChannelUtils.getSessionBy(channel);
36.         messageDispatcher.onSessionClosed(userSession);
37.     }
39. }

复制代码

必要注意的是，UDP Socket Server的链接建立，只在启动的时间触发一次。之后，无论有多少个客户端，上文的channelActive都不会再次触发。也就是说，客户端的链接不是一对一的，全局只有一个服务端链接。这点与tcp不同。那么，怎么区分不同的客户端呢？后文例子发表。
4.2、消息路由

具体的消息处理（通过消息路由及消息处理器注解）。网关担当到消息之后，自动把消息分发到对应的处理器。类似与springmvc的Controller以及RequestMapper功能。

1. public class MessageIoDispatcher extends ChainedMessageDispatcher {
3.     private MessageHandlerRegister handlerRegister;
5.     MessageFactory messageFactory = GameMessageFactory.getInstance();
7.     private MessageParameterConverter msgParameterConverter= new DefaultMessageParameterConverter(messageFactory);
9.     public MessageIoDispatcher() {
10.         LoginRouter router = new LoginRouter();
11.         this.handlerRegister = new CommonMessageHandlerRegister(Collections.singletonList(router), messageFactory);
12.         MessageHandler messageHandler = (session, message) -> {
13.             int cmd = GameMessageFactory.getInstance().getMessageId(message.getClass());
14.             MessageExecutor cmdExecutor = handlerRegister.getMessageExecutor(cmd);
15.             if (cmdExecutor == null) {
16.                 logger.error("message executor missed,  cmd={}", cmd);
17.                 return true;
18.             }
20.             Object[] params = msgParameterConverter.convertToMethodParams(session, cmdExecutor.getParams(), message);
21.             Object controller = cmdExecutor.getHandler();
23.             MessageTask task = MessageTask.valueOf(session, session.hashCode(), controller, cmdExecutor.getMethod(), params);
24.             task.setRequest(message);
25.             // 丢到任务消息队列，不在io线程进行业务处理
26.             GameServer.getMonitorGameExecutor().accept(task);
27.             return true;
28.         };
30.         addMessageHandler(messageHandler);
31.     }
33.     @Override
34.     public void onSessionCreated(IdSession session) {
35.     }
37.     @Override
38.     public void onSessionClosed(IdSession session) {
39.     }
41. }

复制代码

4.3、服务端启动代码

1. public class UdpSocketServer implements ServerNode {
3.     private static final Logger logger = LoggerFactory.getLogger("socketserver");
5.     private EventLoopGroup group = new NioEventLoopGroup();
7.     protected HostAndPort nodesConfig = HostAndPort.valueOf(8088);
9.     public SocketIoDispatcher socketIoDispatcher;
11.     public MessageFactory messageFactory;
13.     public MessageCodec messageCodec;
15.     @Override
16.     public void start() throws Exception {
17.         try {
18.             SessionManager.getInstance().schedule();
19.             
20.             Bootstrap bootstrap = new Bootstrap();
21.             bootstrap.group(group)
22.                     .channel(NioDatagramChannel.class)
23.                     .handler(new LoggingHandler(LogLevel.DEBUG))
24.                     .handler(new ChannelInitializer<DatagramChannel>() {
25.                         @Override
26.                         public void initChannel(DatagramChannel ch) throws Exception {
27.                             ChannelPipeline pipeline = ch.pipeline();
28.                             pipeline.addLast("protocolDecoder", new UdpProtocolDecoder(messageFactory, messageCodec));
29.                             pipeline.addLast("protocolEncoder", new UdpProtocolEncoder(messageFactory, messageCodec));
30.                             pipeline.addLast(new UdpChannelIoHandler(socketIoDispatcher));
32.                         }
33.                     });
35.             logger.info("socket server is listening at " + nodesConfig.getPort() + "......");
36.             bootstrap.bind(nodesConfig.getPort()).sync().channel().closeFuture().sync();
38.         } catch (Exception e) {
39.             logger.error("", e);
40.             group.shutdownGracefully();
41.         }
42.     }
44.     @Override
45.     public void shutdown() throws Exception {
46.         group.shutdownGracefully();
47.     }
50.     public static void main(String[] args) throws Exception {
51.         UdpSocketServer udpSocketServer = new UdpSocketServer();
52.         udpSocketServer.messageFactory = GameMessageFactory.getInstance();
53.         udpSocketServer.messageCodec = new StructMessageCodec();
54.         udpSocketServer.socketIoDispatcher = new MessageIoDispatcher();
56.         udpSocketServer.start();
57.     }
58. }

复制代码

5、客户端代码

5.1、客户端启动代码

1. public class UdpSocketClient extends AbstractSocketClient {
3.     private final EventLoopGroup group = new NioEventLoopGroup(1);
5.     private HostAndPort nativeHostPort;
8.     public UdpSocketClient(SocketIoDispatcher messageDispatcher, MessageFactory messageFactory, MessageCodec messageCodec, HostAndPort hostPort) {
9.         this.ioDispatcher = messageDispatcher;
10.         this.messageFactory = messageFactory;
11.         this.messageCodec = messageCodec;
12.         this.targetAddress = hostPort;
13.     }
15.     @Override
16.     public IdSession openSession() throws IOException {
17.         try {
18.             final NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup();
19.             Bootstrap bootstrap = new Bootstrap();
20.             bootstrap.channel(NioDatagramChannel.class);
21.             bootstrap.group(nioEventLoopGroup);
22.             bootstrap.handler(new LoggingHandler(LogLevel.INFO));
23.             bootstrap.handler(new UdpProtoBufClientChannelInitializer());
24.             ChannelFuture f = bootstrap.connect(new InetSocketAddress(targetAddress.getHost(), targetAddress.getPort()),
25.                     new InetSocketAddress(nativeHostPort.getHost(), nativeHostPort.getPort())).sync();
26.             IdSession session = new NSession(f.channel());
27.             this.session = session;
28.             return session;
29.         } catch (Exception e) {
30.             group.shutdownGracefully();
31.             throw new IOException(e);
32.         }
33.     }
35.     @Override
36.     public void close() throws IOException {
37.         this.session.close();
38.     }
41.     public void send(UdpMessage message) {
42.         message.setSenderIp(nativeHostPort.getHost());
43.         message.setSenderPort(nativeHostPort.getPort());
45.         message.setReceiverIp(targetAddress.getHost());
46.         message.setReceiverPort(targetAddress.getPort());
47.         session.send(message);
48.     }
51.     class UdpProtoBufClientChannelInitializer extends ChannelInitializer<NioDatagramChannel> {
52.         @Override
53.         protected void initChannel(NioDatagramChannel ch) throws Exception {
54.             ChannelPipeline pipeline = ch.pipeline();
55.             pipeline.addLast("protocolDecoder", new UdpProtocolDecoder(messageFactory, messageCodec));
56.             pipeline.addLast("protocolEncoder", new UdpProtocolEncoder(messageFactory, messageCodec));
57.             pipeline.addLast(new UdpChannelIoHandler(ioDispatcher));
58.         }
59.     }
60. }

复制代码

这里必要注意下：
客户端必要指定端口与服务器通讯，这样才气方便消息包携带本身的地址信息。
5.2、测试代码

客户端测试代码如下：
模拟10个玩家登录。读者可自行修改程序。udp是不可靠链接，不包管交付。如果在外网跑，是会出现消息丢包，或者乱序。在当地跑，是不太可能出现这种环境。读者可自行测试，同一个角色发送一大堆消息（附加上次序字段），看服务器收到的消息序号是否完备有序。

1. private static AtomicLong idFactory = new AtomicLong(1000);
3.     public static void main(String[] args) throws Exception {
4.         MessageCodec messageCodec = new StructMessageCodec();
5.         GameMessageFactory.getInstance().registeredClassTypes().forEach(Codec::getSerializer);
6.         for (int i = 0; i < 10; i++) {
7.             System.out.println("----------i=" + i);
8.             UdpSocketClient socketClient = new UdpSocketClient(new SocketIoDispatcherAdapter() {
9.                 @Override
10.                 public void dispatch(IdSession session, Object message) {
11.                     System.out.println("receive package ---------" + JsonUtil.object2String(message));
12.                 }
14.             }, GameMessageFactory.getInstance(), messageCodec, HostAndPort.valueOf(8088));
16.             socketClient.nativeHostPort = HostAndPort.valueOf(8099 + i);
17.             socketClient.openSession();
18.             for (int j = 0; j < 1; j++) {
19.                 ReqLogin req = new ReqLogin();
20.                 req.setPlayerId(idFactory.getAndIncrement());
21.                 socketClient.send(req);
22.             }
23.         }
25.     }

复制代码

6、游戏服务器示例功能

6.1、demo逻辑

我们以上面的代码，实现一个简朴的游戏逻辑。

• 玩家根据服务端的地址进行udp通讯。链接建立之后，模拟账号登录逻辑。
  
• 服务器吸取到登录消息之后，立马推送登录乐成的协议。
  
• 服务器每隔一段时间，向全部注册的客户端推送一个消息包。
  

6.2、登录哀求/响应包

1. @MessageMeta(cmd = 55555)
2. public class ReqLogin extends UdpMessage {
4.     private long playerId;
5. }

复制代码

1. @MessageMeta(cmd = 55556)
2. public class ResPlayerLogin extends UdpMessage {
4.     private long playerId;
5. }

复制代码

6.3、登录路由

1. @MessageRoute
2. public class LoginRouter {
4.     @RequestHandler
5.     public void reqTime(IdSession session, ReqLogin req) {
6.         long playerId = req.getPlayerId();
7.         System.out.println("player login" + playerId);
8.         Player player = new Player();
9.         player.setId(playerId);
10.         player.setRemoteAddr(HostAndPort.valueOf(req.getSenderIp(), req.getSenderPort()));
11.         SessionManager.getInstance().register(playerId, player);
12.         ResPlayerLogin resp = new ResPlayerLogin();
13.         resp.setPlayerId(playerId);
14.         player.receive(session, resp);
15.     }
16. }

复制代码

此中SessionManager 类缓存服务器的全局session，以及各个客户端环境的通讯地址。以及，每隔一段时间自动向客户端推送消息。

1. public class SessionManager {
3.     private static SessionManager inst = new SessionManager();
5.     private ConcurrentMap<Long, Player> id2Players = new ConcurrentHashMap<>();
7.     private IdSession serverSession;
9.     public static SessionManager getInstance() {
10.         return inst;
11.     }
13.     public void register(long playerId, Player player) {
14.         id2Players.put(playerId, player);
15.     }
17.     public void buildSession(IdSession session) {
18.         serverSession = session;
19.     }
21.     public void schedule() {
22.         SchedulerManager.getInstance().scheduleAtFixedRate(()->{
23.             id2Players.forEach((key, value) -> {
24.                 ResWelcome push = new ResWelcome();
25.                 push.setTime(System.currentTimeMillis());
26.                 value.receive(serverSession, push);
27.             });
28.         }, TimeUtil.MILLIS_PER_MINUTE, 10*TimeUtil.MILLIS_PER_SECOND);
29.     }
31. }

复制代码

必要注意的是，客户端只有在登录乐成之后，服务器才气绑定玩家与对应的客户端地址，才气自动推送消息。也就是说，在登录之前，服务器也是无法自动推送消息的，在业务上来说，也是没有意义的。
6.4、客户端测试代码输出

7、总结

udp是一种无连接的协议，t提供不可靠性传输。UDP通过尽力交付数据包的方式进行传输，不对数据包的传输状态进行确认和重传，因此速率较快。UDP适用于对实时性要求较高、对数据正确性要求较低的应用场景。比方，如果语音视频服务。如果游戏服务器确实必要使用udp协议的话，必要在应用层办理丢包乱序题目。
对于乱序，担当方可以先把数据都缓存起来，等一段窗口期的数据担当完毕后再分发给业务层。对于丢包，无论是发送方还是担当方，都必要缓存最近发送的消息。以便用于丢包重传。当然，这只是基本的思路，现实处理起来可能会非常复杂。如果考虑到UDP协议的话，可能KCP会是更好的选择。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4