NATS-研究学习

2024-05-31 23:28
文章标签 学习 研究 nats

本文主要是介绍NATS-研究学习，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

NATS-研究学习

文章目录

    • NATS-研究学习
    • @[toc]
      • 介绍说明
      • 提供的服务内容
      • 各模式介绍测试使用
        • 发布订阅（Publish Subscribe）
        • 请求响应（Request Reply）
        • 队列订阅&分享工作（Queue Subscribers & Sharing Work）
        • 小杭写的Demo
      • 简单安装使用与测试
      • JetStream 简单使用Demo
      • Spring 项目整合
      • Nkey 认证连接
      • 参考资料

介绍说明

NATS是一个go语言开发的开源的、轻量、高性能的原生消息系统。消息由主题处理，不依赖于网络位置。它提供了应用程序或服务与底层物理网络之间的抽象层。数据被编码并作为消息，由发布者发送。消息由一个或多个订阅者接收、解码和处理。

NATS使程序可以很容易地跨不同的环境、语言、云提供商和内部系统进行通信。客户机通常通过单个URL连接到NATS系统，然后向主题订阅或发布消息。通过这种简单的设计，NATS允许程序共享通用的消息处理代码，隔离资源和相互依赖。

NATS核心提供最多一次的服务质量。
默认情况下，NATS是一种即发即弃的消息传递系统。

如果订户没有收听主题（没有主题匹配），或者在发送消息时未激活，则不会收到消息。

如果需要高级的东东，可以试用NATS Streaming 进行，属于NATS的一个服务模块了。

**优点：**使用简单，配置简单。速度极快，性能良好。
多语言支持，不依赖于网络位置，client端只需知道nats的节点和约定好的subject名称即可。

**缺点：**对服务器稳定性要求较高，机房出现故障，导致nats server端需要重连。可能需要重启nats-server。
在消息timeout后，需要在reconnection里要重新初始化连接，不方便。

提供的服务内容

NATS支持各种消息传递模型，包括：

发布订阅（Publish Subscribe）
请求回复（Request Reply）
队列订阅（Queue Subscribers )

提供的功能：

纯粹的发布订阅模型（Pure pub-sub）
服务器集群（Cluster mode server）
自动精简订阅者（Auto-pruning of subscribers)
基于文本协议（Text-based protocol）
多服务质量保证（Multiple qualities of service - QoS）

各模式介绍测试使用

		<dependency><groupId>io.nats</groupId><artifactId>jnats</artifactId><version>2.16.13</version></dependency>
发布订阅（Publish Subscribe）

请添加图片描述

NATS将publish/subscribe消息分发模型实现为一对多通信，发布者在 subject 上发送消息，并且监听该Subject在任何活动的订阅者都会收到该消息。

Demo:【测试可用】

//publish
Connection nc = Nats.connect("nats://127.0.0.1:4222");
nc.publish("subject", "hello world".getBytes(StandardCharsets.UTF_8));//subscribe [这个时间内就之后收到一个，就结束了]
Subscription sub = nc.subscribe("subject");
Message msg = sub.nextMessage(Duration.ofMillis(500));
String response = new String(msg.getData(), StandardCharsets.UTF_8);//或者是基于回调的subscribe [这个程序可以保持，持续接收信息]
//subscribe
Dispatcher d = nc.createDispatcher(msg ->{String response = new String(msg.getData(), StandardCharsets.UTF_8);//do something
})
d.subscribe("subject");
请求响应（Request Reply）

请添加图片描述

Request-Reply是现代分布式系统中的常见模式。发布者（crm）发送一个请求，应用程序（ybind，fpga-agent）要么在响应时等待一定的超时，要么异步接收响应。Request()是一个简单方便的API，它提供了一个伪同步的方式，使用了超时timeout设置。它创建了一个收件箱（收件箱是一种subject类型，对请求者唯一），订阅subject，然后发布你的请求消息（消息带reply地址）设置为收件箱的subject，然后等待响应，或者超时取消。

Demo:【测试可用】

// publish 
Connection nc = Nats.connect("nats://127.0.0.1:4222");
String reply = "replyMsg";   // 这个相当于回到的主题
//请求回应方法回调
Dispatcher d = nc.createDispatcher(msg -> {System.out.println("reply: " +  JSON .toJSONString(msg));
}) ;
d.unsubscribe(reply , 1);
//订阅请求
d.subscribe(reply);
//发布请求
nc.publish("requestSub", reply, "request".getBytes(StandardCharsets.UTF_8));//subscribe
Connection nc = Nats.connect("nats://127.0.0.1:4222");
//注册订阅
Dispatcher dispatcher = nc.createDispatcher(msg -> {System.out.println(JSON.toJSONString(msg));nc.publish(msg.getReplyTo(), "this is reply".getBytes(StandardCharsets.UTF_8));
});
dispatcher.subscribe("requestSub");
队列订阅&分享工作（Queue Subscribers & Sharing Work）

NATS提供称为队列订阅的负载均衡功能。

主要功能是将具有相同queue名字的subject进行负载均衡。

请添加图片描述

要创建一个消息队列，订阅者需注册一个队列名。所有的订阅者用同一个队列名，形成一个队列组。当消息发送到主题后，队列组会自动选择一个成员接收消息。尽管队列组有多个订阅者，但每条消息只能被组中的一个订阅者接收。

Demo:【测试可用】

// Subscribe
Connection nc = Nats.connect();
Dispatcher d = nc.createDispatcher(msg -> {//do somethingSystem.out.println("msg: " + new String(msg.getData(),StandardCharsets.UTF_8));
});
d.subscribe("subject", "queName");  //差别就是这个了
小杭写的Demo
/*** 发布Demo*/
public class NatsPublish {public static void main(String[] args) throws IOException, InterruptedException {
//        publishSubscribe();requestReply();}/*** test 请求响应（Request Reply） 模式*/public static void requestReply() throws IOException, InterruptedException {Connection nc = Nats.connect("nats://192.168.137.xxx:4222");// 这个相当于回到的主题String reply = "replyMsg-qingqiuxinxi";//请求回应方法回调Dispatcher d = nc.createDispatcher(msg ->{System.out.println("=========收到返回的信息============");System.out.println("reply:get retuen： " +  JSON.toJSONString(msg));System.out.println( JSON.parseObject(JSON.toJSONString(msg)).get("data") );String data = (String) JSON.parseObject(JSON.toJSONString(msg)).get("data");System.out.println( new String(Base64.decode( data ))  );});d.unsubscribe(reply , 1);//订阅请求d.subscribe(reply);//发布请求System.out.println( "订阅信息："+reply );nc.publish("requestSub", reply, "请求参数，巴拉巴拉1".getBytes(StandardCharsets.UTF_8));// 下面这些用来负载测试的nc.publish("requestSub", reply, "请求参数，巴拉巴拉2".getBytes(StandardCharsets.UTF_8));nc.publish("requestSub", reply, "请求参数，巴拉巴拉3".getBytes(StandardCharsets.UTF_8));nc.publish("requestSub", reply, "请求参数，巴拉巴拉4".getBytes(StandardCharsets.UTF_8));nc.publish("requestSub", reply, "请求参数，巴拉巴拉5".getBytes(StandardCharsets.UTF_8));nc.publish("requestSub", reply, "请求参数，巴拉巴拉6".getBytes(StandardCharsets.UTF_8));}/*** test 发布订阅（Publish Subscribe） 模式*/public static void publishSubscribe() throws IOException, InterruptedException {Connection nc = Nats.connect("nats://192.168.137.xxx:4222");nc.publish("subject", "hello world1111122211111111".getBytes(StandardCharsets.UTF_8));}
}

/*** 订阅Demo*/
public class NatsSubscribe {public static void main(String[] args) throws IOException, InterruptedException {
//        publishSubscribe();requestReply();}/*** test 请求响应（Request Reply） 模式*/public static void requestReply() throws IOException, InterruptedException {//subscribeConnection nc = Nats.connect("nats://192.168.137.xxx:4222");//注册订阅Dispatcher dispatcher = nc.createDispatcher(msg -> {System.out.println("=======收到请求信息===========");System.out.println(JSON.toJSONString(msg));String data = (String) JSON.parseObject(JSON.toJSONString(msg)).get("data");System.out.println( new String(Base64.decode( data ))  );nc.publish(msg.getReplyTo(), "这个是返回的数据，啦啦啦啦啦".getBytes(StandardCharsets.UTF_8));});dispatcher.subscribe("requestSub");// 队列订阅就换成下面这个，负载测试,都启动几个服务，就可以看到接受效果了
//         dispatcher.subscribe("requestSub", "queName");}/*** test 发布订阅（Publish Subscribe） 模式*/public static void publishSubscribe() throws IOException, InterruptedException {Connection nc = Nats.connect("nats://192.168.137.xxx:4222");//        //subscribe [这个时间内就之后收到一个，就结束了]
//        Subscription sub = nc.subscribe("subject");
//        Message msg = sub.nextMessage(Duration.ofMillis(50000));
//        String response = new String(msg.getData(), StandardCharsets.UTF_8);
//        System.out.println(response);//subscribe  [这个程序可以保持，持续接收信息]Dispatcher d = nc.createDispatcher(msg ->{String response = new String(msg.getData(), StandardCharsets.UTF_8);//do somethingSystem.out.println(response);});d.subscribe("subject");}
}

简单安装使用与测试

# 官方安装NATS[单台]
docker pull nats
docker network create nats
docker run --name nats --network nats -p 4222:4222 -p 8222:8222 nats --http_port 8222  -js# 192.168.137.xxx ： 4222
# 然后用，上文中小杭的Demo试试，基础的功能就可以了解了。

JetStream 简单使用Demo

目前这个的Demo使用的是官方的封装例子方法。

结果是，创建流之后，发送数据。消费端接入会获取全部数据。除非消息被删除，否则每次都是全部获取。

当然，正常获取的时候，由于持久化，只要没有删除，消费端都可以请求再次获取的。

// 创建发送流 和 数据 
public static void main(String[] args) throws Exception {jetStream(args);}public static void jetStream(String[] args) throws Exception {ExampleArgs exArgs = ExampleArgs.builder("Publish", args, "").defaultStream("example-stream").defaultSubject("example-subject").defaultMessage("hello").defaultMsgCount(10).defaultServer("nats://192.168.137.xxx:4222").build();String hdrNote = exArgs.hasHeaders() ? ", with " + exArgs.headers.size() + " header(s)" : "";System.out.printf("\nPublishing to %s%s. Server is %s\n\n", exArgs.subject, hdrNote, exArgs.server);try (Connection nc = Nats.connect(ExampleUtils.createExampleOptions(exArgs.server))) {JetStream js = nc.jetStream();// Create the streamNatsJsUtils.createStreamOrUpdateSubjects(nc, exArgs.stream, exArgs.subject);int stop = exArgs.msgCount < 2 ? 2 : exArgs.msgCount + 1;for (int x = 1; x < stop; x++) {// make unique message data if you want more than 1 messageString data = exArgs.msgCount < 2 ? exArgs.message : exArgs.message + "-" + x;// create a typical NATS messageMessage msg = NatsMessage.builder().subject(exArgs.subject).headers(exArgs.headers).data(data, StandardCharsets.UTF_8).build();PublishAck pa = js.publish(msg);System.out.printf("Published message %s on subject %s, stream %s, seqno %d.\n",data, exArgs.subject, pa.getStream(), pa.getSeqno());}}catch (Exception e) {e.printStackTrace();}}

// 消费，并删除流中已处理的数据 
public static void main(String[] args) throws Exception {jetStream();}public static void jetStream() throws IOException, InterruptedException, JetStreamApiException {Connection nc = Nats.connect("nats://192.168.137.xxx:4222");Dispatcher disp = nc.createDispatcher(msg -> {System.out.println("ddddddd"+msg);});JetStream js = nc.jetStream();MessageHandler handler = (msg) -> {// Process the message.// Ack the message depending on the ack modelString response = new String(msg.getData(), StandardCharsets.UTF_8);//do somethingSystem.out.println(response);System.out.println(msg);System.out.println("处理一下数据，然后要删除掉！！");System.out.println(msg.metaData());try {// 处理完数据，要把数据删除掉的，否则会一直在持久队列中。JetStreamManagement jsm = nc.jetStreamManagement();jsm.deleteMessage(msg.metaData().getStream(),msg.metaData().streamSequence());} catch (IOException | JetStreamApiException e) {e.printStackTrace();}};boolean autoAck = true;js.subscribe("example-subject", disp, handler, autoAck);}

一些复杂的功能，还是有需要的时候，再研究一下官方的Demo程序，比文档好理解多了。。。。

Spring 项目整合

参考开源项目：wanlinus/nats-streaming-spring

代码直接打包了；这里就记录一下使用。

// pom <dependency><groupId>io.nats</groupId><artifactId>jnats</artifactId><version>2.16.13</version><scope>compile</scope></dependency>// 配置
spring:nats:natsUrls: nats://192.168.137.xxx:4222// 启动类
@EnableNats
@SpringBootApplication
public class AppApplication {// 测试类
@Component
@RestController
@RequestMapping("/test")
public class TestController extends BaseController {@Autowiredprivate Connection cconnection;@GetMapping("/test")public String test(HttpServletRequest request){String msg = "send msg " + DateUtil.now();// 测试发送普通消息cconnection.publish("xixi", msg.getBytes(StandardCharsets.UTF_8));return "test-success";}/*** 接收 JetStream 的消息* @param message*/@Subscribe(value="haha",type = "JetStream")public void message1(Message message) {System.out.println("接收 JetStream 的消息，进行处理。。。。。。");System.out.println(message);System.out.println(message.getSubject() + " : " + new String(message.getData()));}/*** 接收普通消息* @param message*/@Subscribe(value="xixi")public void message2(Message message) {System.out.println("接收普通消息，进行处理。。。。。。");System.out.println(message);System.out.println(message.getSubject() + " : " + new String(message.getData()));}
}

其他类型的封装 和 发送操作，就真实需要的时候再继续完善一下了。

Nkey 认证连接

AuthHandler authHandler = Nats.staticCredentials("UCVU4OEHWAxxxxxxxxxxxxDDIxxxxxBMYxxxxxxxxxxxxxxxxxx".toCharArray(),"SUAMMIOB6xxxxxxxxxxxxxxxxxSHYxxxx7MUxxxxxxxxxxx5FCI".toCharArray());Options.Builder builder = new Options.Builder()// 配置 nats 服务器地址.servers(new String[]{"nats://xxxx.xxxx.xxx:4222"}).authHandler(authHandler);Connection nc = Nats.connect(builder.build());

参考资料

  • 简单看看：https://www.jianshu.com/p/341082dadd3e
  • 详细点说明：http://www.guoxiaolong.cn/blog/?id=10376
  • JetStream：https://docs.nats.io/nats-concepts/jetstream
  • Developing With NATS：https://docs.nats.io/using-nats/developer
  • https://docs.nats.io/running-a-nats-service/nats_docker/nats-docker-tutorial
  • 官方javademo：https://github.com/nats-io/nats.java 参考这个 【这个重点的样子】
  • 发布订阅：https://blog.csdn.net/qq_47848696/article/details/117746807
  • https://zhuanlan.zhihu.com/p/628371358 用户+密码连接

这篇关于NATS-研究学习的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


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

相关文章

HarmonyOS学习(七)——UI（五）常用布局总结

HarmonyOS学习(七)——UI（五）常用布局总结

自适应布局 1.1、线性布局（LinearLayout） 通过线性容器Row和Column实现线性布局。Column容器内的子组件按照垂直方向排列，Row组件中的子组件按照水平方向排列。 属性说明space通过space参数设置主轴上子组件的间距，达到各子组件在排列上的等间距效果alignItems设置子组件在交叉轴上的对齐方式，且在各类尺寸屏幕上表现一致，其中交叉轴为垂直时，取值为Vert
阅读更多...
Ilya-AI分享的他在OpenAI学习到的15个提示工程技巧

Ilya-AI分享的他在OpenAI学习到的15个提示工程技巧

Ilya（不是本人，claude AI）在社交媒体上分享了他在OpenAI学习到的15个Prompt撰写技巧。 以下是详细的内容： 提示精确化:在编写提示时,力求表达清晰准确。清楚地阐述任务需求和概念定义至关重要。例:不用"分析文本",而用"判断这段话的情感倾向:积极、消极还是中性"。 快速迭代:善于快速连续调整提示。熟练的提示工程师能够灵活地进行多轮优化。例:从"总结文章"到"用
阅读更多...
【前端学习】AntV G6-08 深入图形与图形分组、自定义节点、节点动画（下）

【前端学习】AntV G6-08 深入图形与图形分组、自定义节点、节点动画（下）

【课程链接】 AntV G6：深入图形与图形分组、自定义节点、节点动画（下）_哔哩哔哩_bilibili 本章十吾老师讲解了一个复杂的自定义节点中，应该怎样去计算和绘制图形，如何给一个图形制作不间断的动画，以及在鼠标事件之后产生动画。（有点难，需要好好理解） <!DOCTYPE html><html><head><meta charset="UTF-8"><title>06
阅读更多...
学习hash总结

学习hash总结

2014/1/29/   最近刚开始学hash，名字很陌生，但是hash的思想却很熟悉，以前早就做过此类的题，但是不知道这就是hash思想而已，说白了hash就是一个映射，往往灵活利用数组的下标来实现算法，hash的作用：1、判重；2、统计次数；
阅读更多...
零基础学习Redis(10) -- zset类型命令使用

零基础学习Redis(10) -- zset类型命令使用

zset是有序集合，内部除了存储元素外，还会存储一个score，存储在zset中的元素会按照score的大小升序排列，不同元素的score可以重复，score相同的元素会按照元素的字典序排列。 1. zset常用命令 1.1 zadd  zadd key [NX | XX] [GT | LT]   [CH] [INCR] score member [score member ...]
阅读更多...
【机器学习】高斯过程的基本概念和应用领域以及在python中的实例

【机器学习】高斯过程的基本概念和应用领域以及在python中的实例

引言 高斯过程（Gaussian Process，简称GP）是一种概率模型，用于描述一组随机变量的联合概率分布，其中任何一个有限维度的子集都具有高斯分布 文章目录 引言一、高斯过程1.1 基本定义1.1.1 随机过程1.1.2 高斯分布 1.2 高斯过程的特性1.2.1 联合高斯性1.2.2 均值函数1.2.3 协方差函数（或核函数） 1.3 核函数1.4 高斯过程回归（Gauss
阅读更多...
【学习笔记】 陈强-机器学习-Python-Ch15 人工神经网络（1）sklearn

【学习笔记】 陈强-机器学习-Python-Ch15 人工神经网络（1）sklearn

系列文章目录 监督学习：参数方法 【学习笔记】 陈强-机器学习-Python-Ch4 线性回归 【学习笔记】 陈强-机器学习-Python-Ch5 逻辑回归 【课后题练习】 陈强-机器学习-Python-Ch5 逻辑回归（SAheart.csv） 【学习笔记】 陈强-机器学习-Python-Ch6 多项逻辑回归 【学习笔记 及 课后题练习】 陈强-机器学习-Python-Ch7 判别分析 【学
阅读更多...
系统架构师考试学习笔记第三篇——架构设计高级知识（20）通信系统架构设计理论与实践

系统架构师考试学习笔记第三篇——架构设计高级知识（20）通信系统架构设计理论与实践

本章知识考点：         第20课时主要学习通信系统架构设计的理论和工作中的实践。根据新版考试大纲,本课时知识点会涉及案例分析题(25分),而在历年考试中,案例题对该部分内容的考查并不多,虽在综合知识选择题目中经常考查,但分值也不高。本课时内容侧重于对知识点的记忆和理解,按照以往的出题规律,通信系统架构设计基础知识点多来源于教材内的基础网络设备、网络架构和教材外最新时事热点技术。本课时知识
阅读更多...
线性代数|机器学习-P36在图中找聚类

线性代数|机器学习-P36在图中找聚类

文章目录 1. 常见图结构2. 谱聚类 感觉后面几节课的内容跨越太大，需要补充太多的知识点，教授讲得内容跨越较大，一般一节课的内容是书本上的一章节内容，所以看视频比较吃力，需要先预习课本内容后才能够很好的理解教授讲解的知识点。 1. 常见图结构 假设我们有如下图结构： Adjacency Matrix:行和列表示的是节点的位置，A[i,j]表示的第 i 个节点和第 j 个
阅读更多...
Node.js学习记录（二）

Node.js学习记录（二）

目录 一、express 1、初识express 2、安装express 3、创建并启动web服务器 4、监听 GET&POST 请求、响应内容给客户端 5、获取URL中携带的查询参数 6、获取URL中动态参数 7、静态资源托管 二、工具nodemon 三、express路由 1、express中路由 2、路由的匹配 3、路由模块化 4、路由模块添加前缀 四、中间件
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2