ZeroMQ-学习

来源：oschina

链接：https://my.oschina.net/u/167092/blog/3211827

ZeroMQ简介

ZeroMQ（也称为ØMQ，0MQ或ZMQ）是一种高性能的异步消息传递库，旨在用于分布式或并行应用程序中。它提供了一个消息队列，但是与面向消息的中间件不同，ZeroMQ系统可以在没有专用消息代理的情况下运行。

ZeroMQ通过各种传输（TCP，进程内，进程间，多播，WebSocket等）支持通用消息传递模式（发布/订阅，请求/答复，客户端/服务器等），使进程间消息传递变得简单作为线程间消息传递。这样可以使您的代码清晰，模块化并且易于扩展。

ZeroMQ由大型贡献者社区开发。许多流行的编程语言都有第三方绑定，而C＃和Java有本地端口。

ZeroMQ 特点

ZeroMQ的理念从零开始。零表示零代理（ZeroMQ是无代理），零延迟，零成本（免费）和零管理。

更广泛地说，“零”是指渗透到项目中的极简主义文化。我们通过消除复杂性而不是通过公开新功能来增加功能。

Socket API

Socket是用于网络编程的事实上的标准API。这就是ZeroMQ提供熟悉的基于Socket的API的原因。使ZeroMQ特别受开发人员欢迎的一件事是，它使用不同的Socket类型来实现任何任意的消息传递模式。此外，ZeroMQ Socket提供了对基础网络协议的干净抽象，从而隐藏了那些协议的复杂性，并使它们之间的切换变得非常容易。

与传统Socket的主要区别

一般而言，常规Socket提供了到面向连接的可靠字节流（SOCK_STREAM）或无连接不可靠数据报（SOCK_DGRAM）的同步接口。相比之下，ZeroMQ Socket提供了异步消息队列的抽象，其确切的排队语义取决于所使用的Socket类型。在常规Socket传输字节流或离散数据报流的地方，ZeroMQ Socket传输离散消息。

ZeroMQ Socket是异步的，这意味着建立连接和关闭，重新连接以及有效交付的时间对用户是透明的，并且由ZeroMQ本身进行组织。此外，在对等端不可接收消息的情况下，消息可以排队。

传统Socket仅允许严格的一对一（两个对等方），多对一（许多客户端，一台服务器），或者在某些情况下允许一对多（多播）关系。除PAIR Socket外，ZeroMQ Socket可以连接到多个端点，同时接受来自绑定到该套接字的多个端点的传入连接，从而允许多对多关系。

Socket 生命周期

ZeroMQ Socket四个部分，就像BSD Socket一样：

创建和销毁Socket，形成了socket 生命周期。
通过在Socket上设置选项并在必要时进行检查来配置Socket。
通过创建与Socket之间的ZeroMQ连接，将Socket插入网络拓扑。
使用Socket通过在Socket上写入和接收消息来承载数据。

绑定与连接

使用ZeroMQ Socket，谁绑定和谁连接都没有关系。服务器使用Bind而客户端使用Connect。为什么会这样，有什么区别？

ZeroMQ为每个基础连接创建队列。如果您的Socket连接到三个对等Socket，则在后台有三个消息队列。

使用“绑定”，可以允许对等方连接到您，因此您不知道将来会有多少个对等方，因此无法提前创建队列。而是在各个对等方连接到绑定的Socket时创建队列。

通过Connect，ZeroMQ知道将至少有一个对等端，因此它可以立即创建一个队列。这适用于ROUTER以外的所有Socket类型，其中ROUTER仅在我们连接的对等方确认我们的连接之后才创建队列。

因此，在将消息发送到没有对等端的绑定Socket或没有实时连接的ROUTER时，没有队列可以存储消息。

什么时候应该使用bind和什么时候用connect ？

通常，请使用体系结构中最稳定的点进行绑定，并使用具有易失性端点的动态组件进行连接。对于请求/答复，服务提供商可能是您绑定和客户端正在使用connect的地方。就像普通的旧TCP。

如果您不能确定哪个部分更稳定（即对等），请考虑在中间安装一个稳定的设备，各方可以连接到该设备。

ZMQ 的四个基本模型

内置的核心ZeroMQ模式是：

Request-reply，它将一组客户端连接到一组服务。这是一个远程过程调用和任务分配模式。
Pub-sub，它将一组发布者连接到一组订阅者。这是一种数据分发模式。
Pipeline，以扇出/扇入模式连接节点，该模式可以具有多个步骤和循环。这是并行的任务分配和收集模式。
Exclusive pair, 专门连接两个Socket。这是用于在一个进程中连接两个线程的模式。

请求-回复模式

请求-应答模式旨在用于各种面向服务的体系结构。它有两种基本类型：同步（REQ和REP 套接字类型）和异步套接字类型（DEALER和ROUTER套接字类型），它们可以以多种方式混合。

请求-应答模式由RFC 28 / REQREP正式定义。

REQ

一个REQ用于由客户端从服务请求发送和接收回复。此Socket类型仅允许发送和后续接收调用的交替序列。一个REQSocket可以连接到任意数量的 REP或ROUTERSocket。发送的每个请求都在所有已连接的服务之间进行轮询，并且收到的每个答复都与上一个发出的请求匹配。它设计用于简单的请求-答复模型，其中对失败的对等节点的可靠性不成问题。

如果没有服务可用，则Socket上的任何发送操作都将阻塞，直到至少一项服务可用为止。该REQ插座不会丢弃任何消息。

REP

一个REPSocket用于由服务接收来自请求和发送回复给客户端。此套接字类型仅允许接收和后续发送调用的交替序列。接收到的每个请求都将在所有客户端之间公平排队，并且每个发送的答复都将路由到发出最后一个请求的客户端。如果原始请求者不再存在，则答复将被静默丢弃。

DEALER socket

ROUTERSocket 会谈到一组对等体，使用显式寻址，使得每个传出消息被发送到特定的对等连接。ROUTER 用作的异步替代REP，并且通常用作与DEALER客户端通信的服务器的基础。

当接收消息时，ROUTERSocket将在消息部分之前包含消息的始发对等方的路由ID，然后再将其传递给应用程序。接收到的消息在所有连接的同级之间公平排队。发送消息时，ROUTERSocket将删除消息的第一部分，并使用它来确定消息应路由到的对等方的路由ID。如果对等点已不存在或不再存在，则该消息将被静默丢弃。

当ROUTERSocket由于已达到所有对等方的最高水位而进入静音状态时，发送到该套接字的任何消息都将被丢弃，直到静音状态结束为止。同样，任何路由到已达到单个高水位标记的对等方的消息也将被丢弃。

当REQSocket连接到ROUTERSocket时，除了始发对等方的 路由ID之外，收到的每个消息还应包含一个空的定界符消息部分。因此，由应用程序看到的每个接收到的消息的整个结构变为：一个或多个路由ID部分，定界符部分，一个或多个主体部分。在向REQ Socket发送答复时，应用程序必须包括定界符部分。

代码示例：

我这里使用的是官网推荐的 JeroMQ 地址 https://github.com/zeromq/jeromq

导入的maven

 <dependency>
      <groupId>org.zeromq</groupId>
      <artifactId>jeromq</artifactId>
      <version>0.5.1</version>
    </dependency>

    <!-- for the latest SNAPSHOT -->
    <dependency>
      <groupId>org.zeromq</groupId>
      <artifactId>jeromq</artifactId>
      <version>0.5.2-SNAPSHOT</version>
    </dependency>

    <!-- If you can't find the latest snapshot -->
    <repositories>
      <repository>
        <id>sonatype-nexus-snapshots</id>
        <url>https://oss.sonatype.org/content/repositories/snapshots</url>
        <releases>
          <enabled>false</enabled>
        </releases>
        <snapshots>
          <enabled>true</enabled>
        </snapshots>
       </repository>
    </repositories>

首先创建一个broker:

package ndm2.ndm2;

import org.zeromq.SocketType;
import org.zeromq.ZContext;
import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Poller;
import org.zeromq.ZMQ.Socket;

public class ZMQDemoBroker {
    
    public static void main(String[] args) {
        try (ZContext context = new ZContext()) {
            Socket frontend = context.createSocket(SocketType.ROUTER);
            Socket backend = context.createSocket(SocketType.DEALER);
            frontend.bind("tcp://127.0.0.1:8700");
            backend.bind("tcp://127.0.0.1:8701");
            System.out.println("broker start");
            Poller items = context.createPoller(2);
            items.register(frontend, Poller.POLLIN);
            items.register(backend, Poller.POLLIN);
            boolean more = false;
            byte[] message;
            while (!Thread.currentThread().isInterrupted()) {
                items.poll();
                if (items.pollin(0)) {
                    while (true) {
                        message = frontend.recv(0);
                        more = frontend.hasReceiveMore();
                        backend.send(message, more ? ZMQ.SNDMORE : 0);
                        if (!more) {
                            break;
                        }
                    }
                }
                if (items.pollin(1)) {
                    while (true) {
                        message = backend.recv(0);
                        more = backend.hasReceiveMore();
                        frontend.send(message, more ? ZMQ.SNDMORE : 0);
                        if (!more) {
                            break;
                        }
                    }
                }
            }
        }
    }
}

然后client 端：

package ndm2.ndm2;

import org.zeromq.SocketType;
import org.zeromq.ZContext;
import org.zeromq.ZMQ.Socket;

public class ZMQDemoClient {
    
    public static void main(String[] args) {
        ZContext context = new ZContext();
        Socket requester = context.createSocket(SocketType.REQ);
        requester.connect("tcp://localhost:8700");
        System.out.println("Client链接成功....");
        for (int request_nbr = 0; request_nbr < 10; request_nbr++) {
            requester.send("你好", 0);
            String reply = requester.recvStr(0);
            System.out.println("收到回复  [" + reply + "]");
        }        
                
    }

}

然后worker

package ndm2.ndm2;

import org.zeromq.SocketType;
import org.zeromq.ZContext;
import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Socket;

public class ZMQDemoWorker {
    
    
    public static void main(String[] args) throws InterruptedException {
        ZContext context = new ZContext();
        Socket responder = context.createSocket(SocketType.REP);
        responder.connect("tcp://localhost:8701");
        System.out.println("Worker链接成功....");
        while (!Thread.currentThread().isInterrupted()) {
              String string = responder.recvStr(0);
              System.out.printf("收到回复: [%s]\n", string);
              Thread.sleep(1000);
              responder.send("World");
        }
    }

}

发布-订阅模式

发布-订阅模式用于以扇出方式将数据从单个发布者一对多分发到多个订阅者。

发布-订阅模式由RFC 29 / PUBSUB正式定义。

ZeroMQ通过以下四种Socket类型支持发布/订阅：

PUB 类型
XPUB 类型
SUB 类型
XSUB 类型

Topics

ZeroMQ使用多部分消息来传达主题信息。主题表示为字节数组，尽管您可以使用字符串并使用适当的文本编码。

发布者必须在邮件有效载荷之前在邮件的第一帧中包含主题。例如，要向状态主题的订户发布状态消息：一个订户Socket可以具有多个订阅筛选器。

使用前缀检查将消息的主题与订户的订阅主题进行比较。

也就是说，订阅的订户topic将收到带有以下主题的消息：

topic
topic/subtopic
topical

但是，它将不会收到带有以下主题的消息：

topi
TOPIC （请记住，这是按字节比较）

PUB

PUB用于通过发布者分发数据。发送的消息以扇出方式分发给所有连接的对等方。此套接字类型无法接收任何消息。

当PUB由于已达到订户的最高水位而进入静音状态时，将发送给有问题的订户的任何消息都将被丢弃，直到静音状态结束为止。send函数永远不会阻塞此套接字类型。

特征


兼容的对等套接字	`SUB`， `XSUB`
方向	单向
发送/接收模式	仅发送
入网路由策略	不适用
外发路由策略	扇出
静音状态下的动作	下降

SUB

SUB用于由订户订阅由发布者分发的数据。


兼容的对等套接字	`PUB`， `XPUB`
方向	单向
发送/接收模式	仅接收
入网路由策略	公平排队
外发路由策略	不适用

XPUB

与PUB相同，只是可以以传入消息的形式接收来自对等方的订阅。订阅消息是字节1（用于订阅）或字节0（用于取消订阅），后跟订阅主体。也接收没有子/取消订阅前缀的消息，但对订阅状态没有影响。

特征摘要：


兼容的对等套接字	ZMQ_SUB，ZMQ_XSUB
方向	单向
发送/接收模式	发送消息，接收订阅
入网路由策略	不适用
外发路由策略	扇出
静音状态下的动作	下降

XSUB

与SUB相同，只是通过向套接字发送订阅消息来进行订阅。订阅消息是字节1（用于订阅）或字节0（用于取消订阅），后跟订阅主体。也可以发送不带sub / unsub前缀的消息，但对订阅状态没有影响。

特征摘要：


兼容的对等套接字	ZMQ_PUB，ZMQ_XPUB
方向	单向
发送/接收模式	接收消息，发送订阅
入网路由策略	公平排队
外发路由策略	不适用
静音状态下的动作	下降

SUB/PUB 示例 这种模式不需要有broker

SUB端

public class ZMQDemoSub {
    
    
    public static void main(String[] args) throws InterruptedException {
        ZContext context = new ZContext();
        Socket subscriber = context.createSocket(SocketType.SUB);
        subscriber.connect("tcp://localhost:5556");
        subscriber.subscribe("test".getBytes(ZMQ.CHARSET));
        while (true){
            String topic = subscriber.recvStr();
            System.out.println("主题为:"+topic);
            String data = subscriber.recvStr();
           System.out.println("消息为:"+data);
       }
    }
}

pub 端

public static void main(String[] args) throws InterruptedException {
         ZContext context = new ZContext();
         Socket publisher = context.createSocket(SocketType.PUB);
         publisher.bind("tcp://127.0.0.1:5556");
         int topicNbr;
         for (topicNbr = 0; topicNbr < 1000; topicNbr++) {
             publisher.send("test", ZMQ.SNDMORE);
             publisher.send("这是一个测试");
             Thread.sleep(1000);
         }
        
    }

posted @ 2020-12-12 22:12 快乐的额阅读(389) 评论(0) 编辑收藏举报

会员力量，点亮园子希望

刷新页面返回顶部