基于Netty实现可靠消息传递的重发机制详解

2024-06-20 07:20
文章标签 消息传递 netty 重发 详解 可靠 机制 实现

本文主要是介绍基于Netty实现可靠消息传递的重发机制详解，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

基于Netty实现可靠消息传递的重发机制详解

本文详细介绍了如何使用Netty框架实现可靠的消息传递机制，特别是消息的重发机制。Netty本身没有内置重发功能，但通过定时任务、消息确认和重试策略，我们可以构建一个健壮的重发系统。示例代码包括客户端和服务器端的实现，展示了如何在发送消息失败或未收到确认时进行重发，确保消息可靠传递。这一机制对于需要高可靠性的数据传输应用非常有用。

基本思路

  • 消息发送和重发逻辑：每次发送消息时，记录该消息以及发送时间，并在一定时间内等待响应。如果没有响应，则重新发送该消息，直到达到最大重发次数。
  • 消息确认：服务器接收到消息后，需要返回一个确认消息（ACK），客户端收到ACK后可以认为该消息发送成功。
  • 超时检测：使用定时任务来检测消息是否超时，如果超时则重发。

代码示例

以下是一个实现上述思路的详细代码示例：

1. 客户端代码

首先，定义一个Netty客户端，包含重发机制。

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;public class NettyClient {private final String host;private final int port;private final Bootstrap bootstrap;private final EventLoopGroup group;private final ConcurrentHashMap<String, Message> pendingMessages;public NettyClient(String host, int port) {this.host = host;this.port = port;this.group = new NioEventLoopGroup();this.bootstrap = new Bootstrap();this.pendingMessages = new ConcurrentHashMap<>();}public void start() {try {bootstrap.group(group).channel(NioSocketChannel.class).option(ChannelOption.SO_KEEPALIVE, true).handler(new ChannelInitializer<SocketChannel>() {@Overridepublic void initChannel(SocketChannel ch) {ch.pipeline().addLast(new ClientHandler(pendingMessages));}});ChannelFuture future = bootstrap.connect(host, port).sync();future.channel().closeFuture().sync();} catch (Exception e) {e.printStackTrace();} finally {group.shutdownGracefully();}}public void sendMessage(Channel channel, String message) {Message msg = new Message(message, channel);pendingMessages.put(message, msg);channel.writeAndFlush(message);scheduleResend(msg);}private void scheduleResend(Message msg) {ScheduledFuture<?> future = group.schedule(() -> {if (msg.incrementRetryCount() > 3) {System.err.println("Message failed after 3 retries: " + msg.getContent());pendingMessages.remove(msg.getContent());} else {System.out.println("Resending message: " + msg.getContent());msg.getChannel().writeAndFlush(msg.getContent());scheduleResend(msg);}}, 5, TimeUnit.SECONDS);msg.setFuture(future);}public static void main(String[] args) {NettyClient client = new NettyClient("localhost", 8080);client.start();}
}

2. Message 类

Message类用于封装消息和相关的重发信息。

import io.netty.channel.Channel;import java.util.concurrent.ScheduledFuture;public class Message {private final String content;private final Channel channel;private int retryCount;private ScheduledFuture<?> future;public Message(String content, Channel channel) {this.content = content;this.channel = channel;this.retryCount = 0;}public String getContent() {return content;}public Channel getChannel() {return channel;}public int incrementRetryCount() {return ++retryCount;}public void setFuture(ScheduledFuture<?> future) {this.future = future;}public void cancelFuture() {if (future != null) {future.cancel(true);}}
}

3. ClientHandler 类

ClientHandler处理服务器响应和确认消息。

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;import java.util.concurrent.ConcurrentHashMap;public class ClientHandler extends ChannelInboundHandlerAdapter {private final ConcurrentHashMap<String, Message> pendingMessages;public ClientHandler(ConcurrentHashMap<String, Message> pendingMessages) {this.pendingMessages = pendingMessages;}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {String response = (String) msg;if (pendingMessages.containsKey(response)) {Message message = pendingMessages.remove(response);message.cancelFuture();System.out.println("Received ACK for message: " + response);} else {System.out.println("Received message from server: " + response);}}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {cause.printStackTrace();ctx.close();}
}

4. 服务器代码

服务器简单地返回ACK消息，确认收到客户端的消息。

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;public class NettyServer {private final int port;public NettyServer(int port) {this.port = port;}public void start() {EventLoopGroup bossGroup = new NioEventLoopGroup(1);EventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap bootstrap = new ServerBootstrap();bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {@Overridepublic void initChannel(SocketChannel ch) {ch.pipeline().addLast(new ServerHandler());}});ChannelFuture future = bootstrap.bind(port).sync();future.channel().closeFuture().sync();} catch (Exception e) {e.printStackTrace();} finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}public static void main(String[] args) {NettyServer server = new NettyServer(8080);server.start();}
}

5. ServerHandler 类

ServerHandler处理客户端消息并发送ACK。

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;public class ServerHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {String message = (String) msg;System.out.println("Received message from client: " + message);ctx.writeAndFlush(message);  // Send ACK}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {cause.printStackTrace();ctx.close();}
}

解释

  • 客户端启动：NettyClient启动并连接到服务器。
  • 消息发送和重发：通过sendMessage方法发送消息，并在消息未确认时进行重发。重发的逻辑通过ScheduledFuture实现，每次重发后会重新计划下一次重发，直到达到最大重发次数。
  • 消息确认：客户端在接收到服务器的ACK消息后，取消重发计划并移除待确认的消息。
  • 服务器处理：NettyServer和ServerHandler处理客户端的消息，并简单地返回ACK确认消息。

通过这种方式，我们实现了一个基于Netty的简单消息重发机制。可以根据实际需求进一步扩展和优化，例如添加更多的错误处理、日志记录和不同类型的消息处理。

这篇关于基于Netty实现可靠消息传递的重发机制详解的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1077467。

相关文章

C++对象布局及多态实现探索之内存布局（整理的很多链接）

C++对象布局及多态实现探索之内存布局（整理的很多链接）

本文通过观察对象的内存布局，跟踪函数调用的汇编代码。分析了C++对象内存的布局情况，虚函数的执行方式，以及虚继承，等等 文章链接：http://dev.yesky.com/254/2191254.shtml      论C/C++函数间动态内存的传递 (2005-07-30)   当你涉及到C/C++的核心编程的时候，你会无止境地与内存管理打交道。 文章链接：http://dev.yesky
阅读更多...
十四、观察者模式与访问者模式详解

十四、观察者模式与访问者模式详解

21.观察者模式 21.1.课程目标 1、 掌握观察者模式和访问者模式的应用场景。 2、 掌握观察者模式在具体业务场景中的应用。 3、 了解访问者模式的双分派。 4、 观察者模式和访问者模式的优、缺点。 21.2.内容定位 1、 有 Swing开发经验的人群更容易理解观察者模式。 2、 访问者模式被称为最复杂的设计模式。 21.3.观察者模式 观 察 者 模 式 ( Obser
阅读更多...
通过SSH隧道实现通过远程服务器上外网

通过SSH隧道实现通过远程服务器上外网

搭建隧道 autossh -M 0 -f -D 1080 -C -N user1@remotehost##验证隧道是否生效，查看1080端口是否启动netstat -tuln | grep 1080## 测试ssh 隧道是否生效curl -x socks5h://127.0.0.1:1080 -I http://www.github.com 将autossh 设置为服务，隧道开机启动
阅读更多...
【操作系统】信号Signal超详解|捕捉函数

【操作系统】信号Signal超详解|捕捉函数

🔥博客主页： 我要成为C++领域大神🎥系列专栏：【C++核心编程】 【计算机网络】 【Linux编程】 【操作系统】 ❤️感谢大家点赞👍收藏⭐评论✍️ 本博客致力于知识分享，与更多的人进行学习交流 ​ 如何触发信号 信号是Linux下的经典技术，一般操作系统利用信号杀死违规进程，典型进程干预手段，信号除了杀死进程外也可以挂起进程 kill -l 查看系统支持的信号
阅读更多...
SpringBoot集成Netty，Handler中@Autowired注解为空

SpringBoot集成Netty，Handler中@Autowired注解为空

最近建了个技术交流群，然后好多小伙伴都问关于Netty的问题，尤其今天的问题最特殊，功能大概是要在Netty接收消息时把数据写入数据库，那个小伙伴用的是 Spring Boot + MyBatis + Netty，所以就碰到了Handler中@Autowired注解为空的问题 参考了一些大神的博文，Spring Boot非controller使用@Autowired注解注入为null的问题，得到
阅读更多...
时序预测 | MATLAB实现LSTM时间序列未来多步预测-递归预测

时序预测 | MATLAB实现LSTM时间序列未来多步预测-递归预测

时序预测 | MATLAB实现LSTM时间序列未来多步预测-递归预测 目录 时序预测 | MATLAB实现LSTM时间序列未来多步预测-递归预测基本介绍程序设计参考资料 基本介绍 MATLAB实现LSTM时间序列未来多步预测-递归预测。LSTM是一种含有LSTM区块（blocks）或其他的一种类神经网络，文献或其他资料中LSTM区块可能被描述成智能网络单元，因为
阅读更多...
vue项目集成CanvasEditor实现Word在线编辑器

vue项目集成CanvasEditor实现Word在线编辑器

CanvasEditor实现Word在线编辑器 官网文档：https://hufe.club/canvas-editor-docs/guide/schema.html 源码地址：https://github.com/Hufe921/canvas-editor 前提声明： 由于CanvasEditor目前不支持vue、react 等框架开箱即用版，所以需要我们去Git下载源码，拿到其中两个主
阅读更多...
Jitter Injection详解

Jitter Injection详解

一、定义与作用 Jitter Injection，即抖动注入，是一种在通信系统中人为地添加抖动的技术。该技术通过在发送端对数据包进行延迟和抖动调整，以实现对整个通信系统的时延和抖动的控制。其主要作用包括： 改善传输质量：通过调整数据包的时延和抖动，可以有效地降低误码率，提高数据传输的可靠性。均衡网络负载：通过对不同的数据流进行不同程度的抖动注入，可以实现网络资源的合理分配，提高整体传输效率。增
阅读更多...
android一键分享功能部分实现

android一键分享功能部分实现

为什么叫做部分实现呢，其实是我只实现一部分的分享。如新浪微博，那还有没去实现的是微信分享。还有一部分奇怪的问题：我QQ分享跟QQ空间的分享功能，我都没配置key那些都是原本集成就有的key也可以实现分享，谁清楚的麻烦详解下。 实现分享功能我们可以去www.mob.com这个网站集成。免费的，而且还有短信验证功能。等这分享研究完后就研究下短信验证功能。 开始实现步骤（新浪分享，以下是本人自己实现
阅读更多...
基于Springboot + vue 的抗疫物质管理系统的设计与实现

基于Springboot + vue 的抗疫物质管理系统的设计与实现

目录 📚 前言 📑摘要 📑系统流程 📚 系统架构设计 📚 数据库设计 📚 系统功能的具体实现    💬 系统登录注册 系统登录 登录界面   用户添加  💬 抗疫列表展示模块     区域信息管理 添加物资详情 抗疫物资列表展示 抗疫物资申请 抗疫物资审核 ✒️ 源码实现 💖 源码获取 😁 联系方式 📚 前言 📑博客主页：
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2