# zmq-learn

**Repository Path**: leaning-limitless/zmq-learn

## Basic Information

- **Project Name**: zmq-learn
- **Description**: zmq学习记录
- **Primary Language**: Java
- **License**: Not specified
- **Default Branch**: dev
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 2
- **Forks**: 2
- **Created**: 2019-12-20
- **Last Updated**: 2024-07-29

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# 本介绍基于以下依赖

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

# 参考文档
[ZMQ官方使用指南](http://zguide.zeromq.org/page:all#top)

[ZMQ官方使用指南中文翻译](https://gitee.com/solym/ZeroMQ-Guide-Zh)

[ZMQ官方API文档socket](http://api.zeromq.org/4-0:zmq-socket)

[ZMQ官方API文档socketopt](http://api.zeromq.org/master:zmq-setsockopt)

[ZMQsocketopt中文翻译](https://www.cnblogs.com/fengbohello/p/4398953.html)

[jeromq的源码](https://github.com/booksbyus/zguide)

[jeromq的javadoc](https://www.javadoc.io/doc/org.zeromq/jeromq/latest/org/zeromq/ZMQ.Socket.html)


## 术语
high water mark: 高水位标志，接收或发送消息的缓冲区大小可通过Socket::setRcvHWM和Socket::setSndHWM设置。（各种socket option的含义参考前面给的链接中的文档）

mute state: 静音状态 ，一般是指某个端点无法发送消息的状态，可能是因为没有与之连接的接收方，也可能是因为接收方由于某些原因无法继续接收消息。

# 消息模式

## 请求-回复模式

请求回复模式适用于客户端向服务端发送请求，然后每一个请求都会收到服务端的回复的场景。

[ZMQ rfc/spec28 ](http://rfc.zeromq.org/spec:28)

### 代码样例

https://gitee.com/toosimple-naive/zmq-learn/tree/dev/zmq/src/main/java/com/learn/zmq/repreq

### 包含的stocket type

ZMQ.REQ, ZMQ.REP, ZMQ.DEALER, ZMQ.ROUTER 

#### ZMQ_REQ 

<em>ZMQ_REQ</em> 类型的socket 用于向服务端发送请求并接收回复消息。

只能先发消息后收消息，发出的请求会以round-robin方式分配给不同的服务端节点。

没有服务可用时，发送操作会阻塞，直到至少一个服务可用。

<em>ZMQ_REQ</em> 类型的socket不丢弃消息。

<table class="wiki-content-table">
<tr>
<th colspan="2">ZMQ_REQ 特征</th>
</tr>
<tr>
<td>Compatible peer sockets</td>
<td><em>ZMQ_REP</em>, <em>ZMQ_ROUTER</em></td>
</tr>
<tr>
<td>Direction</td>
<td>Bidirectional</td>
</tr>
<tr>
<td>Send/receive pattern</td>
<td>Send, Receive, Send, Receive, &#8230;</td>
</tr>
<tr>
<td>Outgoing routing strategy</td>
<td>Round-robin</td>
</tr>
<tr>
<td>Incoming routing strategy</td>
<td>Last peer</td>
</tr>
<tr>
<td>Action in mute state</td>
<td>Block</td>
</tr>
</table>

#### ZMQ_REP 

<em>ZMQ_REP </em> 类型的socket 用于接收请求并发送回复的服务。
只能先收消息后发消息。收到的消息以公平队列的方式处理。
当请求方已不存在时，回复消息会被丢弃。

<table class="wiki-content-table">
<tr>
<th colspan="2">ZMQ_REP 特征</th>
</tr>
<tr>
<td>Compatible peer sockets</td>
<td><em>ZMQ_REQ</em>, <em>ZMQ_DEALER</em></td>
</tr>
<tr>
<td>Direction</td>
<td>Bidirectional</td>
</tr>
<tr>
<td>Send/receive pattern</td>
<td>Receive, Send, Receive, Send, &#8230;</td>
</tr>
<tr>
<td>Incoming routing strategy</td>
<td>Fair-queued</td>
</tr>
<tr>
<td>Outgoing routing strategy</td>
<td>Last peer</td>
</tr>
</table>

####  ZMQ_DEALER 

<em>ZMQ_DEALER</em> 类型的socket 是  request/reply sockets 的高级扩展。

所有发出的消息会以round robin的方式发送给连接端点。

连接端点全都达到high water mark时，或者没有连接端点时，进入静音状态(mute state) 。

静音状态下发送操作会被阻塞。

当<em>ZMQ_DEALER</em> socket 连接 到 <em>ZMQ_REP</em> 时，所有发出的消息必须包括：空消息（分隔），后续跟一个或多个body。 

<table class="wiki-content-table">
<tr>
<th colspan="2"> ZMQ_DEALER 特征</th>
</tr>
<tr>
<td>Compatible peer sockets</td>
<td><em>ZMQ_ROUTER</em>, <em>ZMQ_REP</em>, <em>ZMQ_DEALER</em></td>
</tr>
<tr>
<td>Direction</td>
<td>Bidirectional</td>
</tr>
<tr>
<td>Send/receive pattern</td>
<td>Unrestricted</td>
</tr>
<tr>
<td>Outgoing routing strategy</td>
<td>Round-robin</td>
</tr>
<tr>
<td>Incoming routing strategy</td>
<td>Fair-queued</td>
</tr>
<tr>
<td>Action in mute state</td>
<td>Block</td>
</tr>
</table>

#### ZMQ_ROUTER 
<em>ZMQ_ROUTER </em> 类型的socket 是  request/reply sockets 的高级扩展。

接收到消息后会添加一个包含源端点的identity的消息帧，然后再发送给应用程序。

发送消息时会先取出包含目标端点identity的消息帧，以确定应该发送到哪一个端点。

没有端点连接的时消息会被丢弃，除非调用Socket::setRouterMandatory设置ZMQ_ROUTER_MANDATORY为1。

所有连接的端点都达到高水位标志（high water mark）而无法接收消息时，会进入静音状态，此时丢弃所有发来的消息。

当<em>ZMQ_REQ </em>连接到<em>ZMQ_ROUTER </em>时，ZMQ_ROUTER 收到的消息除了包含identity之外，还要包含空消息(分隔标志)。

<table class="wiki-content-table">
<tr>
<th colspan="2">ZMQ_ROUTER 特征</th>
</tr>
<tr>
<td>Compatible peer sockets</td>
<td><em>ZMQ_DEALER</em>, <em>ZMQ_REQ</em>, <em>ZMQ_ROUTER</em></td>
</tr>
<tr>
<td>Direction</td>
<td>Bidirectional</td>
</tr>
<tr>
<td>Send/receive pattern</td>
<td>Unrestricted</td>
</tr>
<tr>
<td>Outgoing routing strategy</td>
<td>See text</td>
</tr>
<tr>
<td>Incoming routing strategy</td>
<td>Fair-queued</td>
</tr>
<tr>
<td>Action in mute state</td>
<td>Drop</td>
</tr>
</table>


## 发布-订阅模式 
发布订阅模式适用于一个发布者向多个订阅者发布消息的场景。

[ZMQ rfc/spec29 ](http://rfc.zeromq.org/spec:29)

### 代码样例
https://gitee.com/toosimple-naive/zmq-learn/tree/dev/zmq/src/main/java/com/learn/zmq/pubsub

### 包含的stocket type
ZMQ.SUB, ZMQ.PUB, ZMQ.XSUB, ZMQ.XPUB

#### ZMQ_PUB 

<em>ZMQ_PUB </em>类型的socket用于发布者分发数据。消息会被发送给所有连接端点，该类型socket只能发送消息，不能接收消息。

<em>ZMQ_PUB </em>类型的socket因为一个订阅者达到高水位标志而进入静音状态时，所有发送给那个出问题的订阅者的消息都会被丢弃，直到静音状态结束。
这种类型的socket的send方法永远不应该阻塞。

<table class="wiki-content-table">
<tr>
<th colspan="2">ZMQ_PUB 特征</th>
</tr>
<tr>
<td>Compatible peer sockets</td>
<td><em>ZMQ_SUB</em>, <em>ZMQ_XSUB</em></td>
</tr>
<tr>
<td>Direction</td>
<td>Unidirectional</td>
</tr>
<tr>
<td>Send/receive pattern</td>
<td>Send only</td>
</tr>
<tr>
<td>Incoming routing strategy</td>
<td>N/A</td>
</tr>
<tr>
<td>Outgoing routing strategy</td>
<td>Fan out</td>
</tr>
<tr>
<td>Action in mute state</td>
<td>Drop</td>
</tr>
</table>

#### ZMQ_SUB

<em>ZMQ_SUB</em>类型的socket用于向发布者订阅消息。Socket::subscribe方法进行订阅。

<table class="wiki-content-table">
<tr>
<th colspan="2">ZMQ_SUB 特征</th>
</tr>
<tr>
<td>Compatible peer sockets</td>
<td><em>ZMQ_PUB</em>, <em>ZMQ_XPUB</em></td>
</tr>
<tr>
<td>Direction</td>
<td>Unidirectional</td>
</tr>
<tr>
<td>Send/receive pattern</td>
<td>Receive only</td>
</tr>
<tr>
<td>Incoming routing strategy</td>
<td>Fair-queued</td>
</tr>
<tr>
<td>Outgoing routing strategy</td>
<td>N/A</td>
</tr>
</table>

#### ZMQ_XPUB
<em>ZMQ_XPUB</em>类型的socket和<em>ZMQ_PUB</em>类型的socket功能基本一致，除了可以消息方式接受订阅。一个订阅消息，以字节1（订阅）或者字节0（取消订阅）开头，后接订阅body。没有前缀的消息也会被接收，但是对订阅状态没有影响。

<table class="wiki-content-table">
<tr>
<th colspan="2">ZMQ_XPUB 特征</th>
</tr>
<tr>
<td>Compatible peer sockets</td>
<td><em>ZMQ_SUB</em>, <em>ZMQ_XSUB</em></td>
</tr>
<tr>
<td>Direction</td>
<td>Unidirectional</td>
</tr>
<tr>
<td>Send/receive pattern</td>
<td>Send messages, receive subscriptions</td>
</tr>
<tr>
<td>Incoming routing strategy</td>
<td>N/A</td>
</tr>
<tr>
<td>Outgoing routing strategy</td>
<td>Fan out</td>
</tr>
<tr>
<td>Action in mute state</td>
<td>Drop</td>
</tr>
</table>

#### ZMQ_XSUB 
<em>ZMQ_XSUB </em>类型的socket和<em>ZMQ_SUB</em>类型的socket功能基本一致，除了以发出消息的方式进行订阅。一个订阅消息，以字节1（订阅）或者字节0（取消订阅）开头，后接订阅body。没有前缀的消息也可以被发出，但是对订阅状态没有影响。

<table class="wiki-content-table">
<tr>
<th colspan="2">ZMQ_XSUB 特征</th>
</tr>
<tr>
<td>Compatible peer sockets</td>
<td><em>ZMQ_PUB</em>, <em>ZMQ_XPUB</em></td>
</tr>
<tr>
<td>Direction</td>
<td>Unidirectional</td>
</tr>
<tr>
<td>Send/receive pattern</td>
<td>Receive messages, send subscriptions</td>
</tr>
<tr>
<td>Incoming routing strategy</td>
<td>Fair-queued</td>
</tr>
<tr>
<td>Outgoing routing strategy</td>
<td>N/A</td>
</tr>
<tr>
<td>Action in mute state</td>
<td>Drop</td>
</tr>
</table>


## 管道模式
管道模式适用于向组织在一个管道中的所有节点发布数据的场景。管道的每一段都连接到至少一个节点，并且当有多个节点连接到同一个管道时，数据是以round-robin方式发送给这些节点的。

[ZMQ rfc/spec30 ](http://rfc.zeromq.org/spec:30)

### 代码样例
https://gitee.com/toosimple-naive/zmq-learn/tree/dev/zmq/src/main/java/com/learn/zmq/pipeline

### 包含的stocket type
 ZMQ.PUSH, ZMQ.PULL

#### ZMQ_PUSH 
<em>ZMQ_PUSH </em>类型的socket用于从一个管道节点向下游管道节点发送消息。消息会以round-robin的方式发送给所有连接到该节点的下游节点。这种类型的socket不能接收消息，只能发送消息。

当<em>ZMQ_PUSH </em>类型的socket因为所有下游节点达到高水位标志时，或者没有下游节点，而进入静音状态时，所有的发送操作都会阻塞，直到有下游节点可接收消息时。消息不会被丢弃。

<table class="wiki-content-table">
<tr>
<th colspan="2">ZMQ_PUSH 特征</th>
</tr>
<tr>
<td>Compatible peer sockets</td>
<td><em>ZMQ_PULL</em></td>
</tr>
<tr>
<td>Direction</td>
<td>Unidirectional</td>
</tr>
<tr>
<td>Send/receive pattern</td>
<td>Send only</td>
</tr>
<tr>
<td>Incoming routing strategy</td>
<td>N/A</td>
</tr>
<tr>
<td>Outgoing routing strategy</td>
<td>Round-robin</td>
</tr>
<tr>
<td>Action in mute state</td>
<td>Block</td>
</tr>
</table>

#### ZMQ_PULL
<em>ZMQ_PULL</em>类型的socket用于接收所有上游管道节点向下游管道节点发送的消息。消息会以fair-queue的方式被接收 。这种类型的socket只能接收消息，不能发送消息。

<table class="wiki-content-table">
<tr>
<th colspan="2">ZMQ_PULL 特征</th>
</tr>
<tr>
<td>Compatible peer sockets</td>
<td><em>ZMQ_PUSH</em></td>
</tr>
<tr>
<td>Direction</td>
<td>Unidirectional</td>
</tr>
<tr>
<td>Send/receive pattern</td>
<td>Receive only</td>
</tr>
<tr>
<td>Incoming routing strategy</td>
<td>Fair-queued</td>
</tr>
<tr>
<td>Outgoing routing strategy</td>
<td>N/A</td>
</tr>
<tr>
<td>Action in mute state</td>
<td>Block</td>
</tr>
</table>

## 排他结对模式

排他结对模式适用于要实现精确的端到端连接的场景。

[ZMQ rfc/spec31 ](http://rfc.zeromq.org/spec:31)

### 代码样例
https://gitee.com/toosimple-naive/zmq-learn/tree/dev/zmq/src/main/java/com/learn/zmq/pair

### 包含的stocket type
 ZMQ.PAIR 

#### ZMQ_PAIR 
<em>ZMQ_PAIR</em>类型的socket每次只能和一个<em>ZMQ_PAIR</em>类型的socket相连接。通过该类型的socket发送的消息，不会进行路由或过滤（因为没有必要）

当<em>ZMQ_PAIR</em>类型的socket因为和它相连接的端点达到高水位标志时，或者没有端点和它连接，而进入静音状态，所有发送操作都会阻塞，直到有可以接收的端点。消息不会被丢弃。

<table class="wiki-content-table">
<tr>
<th colspan="2">ZMQ_PAIR 特征</th>
</tr>
<tr>
<td>Compatible peer sockets</td>
<td><em>ZMQ_PAIR</em></td>
</tr>
<tr>
<td>Direction</td>
<td>Bidirectional</td>
</tr>
<tr>
<td>Send/receive pattern</td>
<td>Unrestricted</td>
</tr>
<tr>
<td>Incoming routing strategy</td>
<td>N/A</td>
</tr>
<tr>
<td>Outgoing routing strategy</td>
<td>N/A</td>
</tr>
<tr>
<td>Action in mute state</td>
<td>Block</td>
</tr>
</table>

## 原生模式
原生模式适用于和TCP端点通信的场景，允许双向的异步请求和回复。

### 代码样例
https://gitee.com/toosimple-naive/zmq-learn/tree/dev/zmq/src/main/java/com/learn/zmq/stream


### 包含的stocket type
ZMQ.STREAM

#### ZMQ_STREAM 

<em>ZMQ_STREAM</em>类型的socket一般用于和非zmq的端点进行交互。当使用tcp协议通信时，该类型的socket可以作为客户端或者服务端，异步发送或接收数据。

接收TCP数据时，<em>ZMQ_STREAM</em>类型的socket会补充一个包含源端点identity的消息，多个端点同时发送给该类型socket的消息被以fair-queue的方式接收。

发送TCP数据时，<em>ZMQ_STREAM</em>类型的socket会移除包含identity的消息，并利用这个identity确定消息发送的目标端点。

所以发送消息前必须先发一个identity frame ，跟着一个实际消息的frame。

断开连接需要先发送一个identity消息frame，跟着一个空消息frame。

<em>ZMQ_STREAM</em>类型的服务端点发送消息时必须使用ZMQ.SNDMORE标志，否则无法继续向其他连接端点发送消息。


<table class="wiki-content-table">
<tr>
<th colspan="2">ZMQ_STREAM 特征</th>
</tr>
<tr>
<td>Compatible peer sockets</td>
<td>none.</td>
</tr>
<tr>
<td>Direction</td>
<td>Bidirectional</td>
</tr>
<tr>
<td>Send/receive pattern</td>
<td>Unrestricted</td>
</tr>
<tr>
<td>Outgoing routing strategy</td>
<td>See text</td>
</tr>
<tr>
<td>Incoming routing strategy</td>
<td>Fair-queued</td>
</tr>
<tr>
<td>Action in mute state</td>
<td>EAGAIN</td>
</tr>
</table>