Nats--java使用nats客户端进行PUB/SUB发送消息

- 日理万妓 2024-04-06 12:24 51阅读 0赞

NATS是一个开源且高性能的消息系统,它常常被认为是”一个为云服务的中央神经系统”.它每秒钟可以传送百万条消息,所以非常适合用来连接微服务和IOT设备。  
NATS是一个发布订阅方式的消息系统。在这类系统中,一个或多个消息发布者将特定的主题发送给一个消息中介者,然后消息中介者再将消息分发给任意客户端(或者这个主题的订阅者)。消息发布者不知道也不关心消息订阅者是谁,反之依然。由于我们可以在不影响系统其它部分的情况下增加新的消息发布者和订阅者,这样的架构使得系统的伸缩性变得很好并且可以比较容易地增加系统的容量。这种类型的系统非常适合用来监测服务器和终端设备;终端设备可以发送消息,我们可以订阅这些消息,然后通过邮件或者其他的方式发送消息通知。

下面我们会部署一个nats-server在windows上，然后使用java创建两个nats客户端，一个发送pub请求,一个发送sub请求，检查sub的这个客户端上能否收到信息。

## 一、服务端部署 ##

1.进入网站https://www.nats.io/download/nats-io/gnatsd/  
2.下载windows版本，也可以用docker-image版，都很方便。  
3.windows版本，下载之后，解压，双击gnatsd.exe即可运行  
4.获得服务器地址：nats://localhost:4222

## 二、nats代码 ##

### 1.先maven安装相关依赖，主要是nats的和json的，因为我们要pub一个json过去 ###

<dependency>
    <groupId>io.nats</groupId>
    <artifactId>jnats</artifactId>
    <version>2.2.0</version>
    </dependency>
    <dependency>
    <groupId>com.google.code.gson</groupId>
    <artifactId>gson</artifactId>
    <version>2.8.5</version>
    </dependency>

### 2.nats客户端(sub订阅) ###

package nats.lzwtest;
    import java.nio.charset.StandardCharsets;
    import java.util.concurrent.CountDownLatch;
    import io.nats.client.Connection;
    import io.nats.client.Dispatcher;
    import io.nats.client.Nats;
    public class SubscribeAsync {
        
        public static void main(String[] args) {
        
            try {
        
                // [begin subscribe_async]
            	Connection nc = Nats.connect("nats://localhost:4222");
                // Use a latch to wait for a message to arrive
                CountDownLatch latch = new CountDownLatch(10);
                // Create a dispatcher and inline message handler
                Dispatcher d = nc.createDispatcher((msg) -> {
        
                    String str = new String(msg.getData(), StandardCharsets.UTF_8);
                    System.out.println(str);
                    latch.countDown();
                });
                // Subscribe
                d.subscribe("lizhenwei");
                // Wait for a message to come in
                latch.await();
                // Close the connection
                nc.close();
                // [end subscribe_async]
            } catch (Exception e) {
        
                e.printStackTrace();
            }
        }
    }

## 3.nats客户端(pub发布) ##

package nats.lzwtest;
    import java.nio.charset.StandardCharsets;
    import java.time.Duration;
    import com.google.gson.Gson;
    import com.google.gson.GsonBuilder;
    import io.nats.client.Connection;
    import io.nats.client.Nats;
    // [begin publish_json]
    class StockForJsonPub {
        
        public String symbol;
        public float price;
    }
    public class PublishJSON {
        
        public static void main(String[] args) {
        
            try {
        
            	Connection nc = Nats.connect("nats://localhost:4222");
                // Create the data object
                StockForJsonPub stk = new StockForJsonPub();
                stk.symbol="GOOG";
                stk.price=1200;
                // use Gson to encode the object to JSON
                GsonBuilder builder = new GsonBuilder();
                Gson gson = builder.create();
                String json = gson.toJson(stk);
                // Publish the message
                nc.publish("lizhenwei", json.getBytes(StandardCharsets.UTF_8));
                // Make sure the message goes through before we close
                nc.flush(Duration.ZERO);
                nc.close();
                System.out.println("pub success");
            } catch (Exception e) {
        
                e.printStackTrace();
            }
        }
    }
    // [end publish_json]

运行结果  
sub端收到如下：

{"symbol":"GOOG","price":1200.0}