ZPER is ZeroMQ persistence broker. ZPER store ZMQ message frames to disk as is.
Inspired by Apache Kafka
- Easier to use for 0MQ developer
- 2x faster than Kafka at Producer (1.5x faster at Consumer)
- No ZooKeepr dependency
- Listen on the ROUTER socket
- Handle write requests
- Multiplex worker thread based on the topic hash
- Support replica to another ZPWriter (at 0.2)
- Listen on the ROUTER socket
- Handle fetch requests
- Handle operation requests
- Listen on the PUB socket for public
- Connected SUB will get last sucessfully synced offset and the first frame of the message
- So try to use the first frame as a message id
- By subscribing, an application get more fined grained offset
- Listen on the inproc PULL socket for ack input from the ZPWriter
- Keep basic stats
- Listen on the ROUTER socket for public
- ZPRelay poll the data file and send new messages to client
- If the ZPER is restarted, ZPRelay starts to send messages from the topic's last offset.
- Offers a seamless experience
- When ZPER works as a cluster, ZPER coordinates through ZPController
- Listen on the ROUTER socket
- ZPController connects each other and shares informaion
- Client can get cluster information
- Keep basic stats
- follows ZMTP 2.0 frame
# if a message has more frame
flag |= 0x01
# when payload length is less than 255
+ ----------- + ------------- + -------------------- +
| 1 byte-flag | 1 byte-length | length-bytes-payload |
+ ----------- + ------------- + -------------------- +
# when payload length exceeds 255
flag |= 0x02
+ ----------- + -------------- + ---------------------+
| 1 byte-flag | 8 bytes-length | length-bytes-payload |
+ ----------- + -------------- + ---------------------+
# common configuration
io_threads = 2 # number of io threads
base_dir = "/data/zper" # data store base directory
segment_size = 536870912 # segment file size 512M
flush_messages = 10000 # flush every N messages
flush_interval = 1000 # flush every N milliseconds
retain_hours = 168 # cleanup segements older than 1 week
# writer configuration
writer.bind = tcp://*:5555 # writer bind address
writer.workers = 3 # number of worker threads
writer.replica.1 = tcp://addr1:port # target replica 1's address
writer.replica.2 = tcp://addr2:port # target replica 2's address
# reader configuration
reader.bind = tcp://*:5556 # reader bind address
reader.workers = 3 # number of worker threads
# ack configuration
ack.bind = tcp://*:5557
# relay configuration
relay.bind = tcp://*:5558
relay.poll = 1000 # poll every N milliseconds
# controller configuration (at 0.2)
controller.bind = tcp://*:5559
controller.quorum.1 = tcp://addr1:port # target quorum 1's address
controller.quorum.2 = tcp://addr2:port # target quorum 2's address
-
Start ZPER ./bin/zper-server-start.sh conf/server.properties
-
Stop ZPER ./bin/zper-server-stop.sh
- Create a DEALER, REQ or PUSH socket and set proper HWM
- Set a identity as the following format (ZMQ 3.2.2 or above is required for PUSH)
- Connect to ZPWriter. To load balance just connect to multiple ZPWriter
+ ----------------- + ----------- + ------------------- + ----- + ------------- +
| 1 byte-topic-hash | 1 byte-mode | 1 byte-topic-length | topic | 16 bytes uuid |
+ ----------------- + ----------- + ------------------- + ----- + ------------- +
Hash function is
0xff & (s[0]*31^(n-1) + s[1]*31^(n-2) + ... + s[n-1])
Flags is
0 (Fastest) : No response (DEALER and PUSH)
1 : Response at flush (DEALER)
2 (Slowest) : Response at every write (REQ)
Context ctx = ZMQ.context (1); Socket sock = ctx.socket (ZMQ.DEALER); sock.setIdentity (ZPUtils.genTopicIdentity (topic, 0)); // mode = 0 sock.connect ("tcp://127.0.0.1:5555"); while (true) { String data = "hello"; sock.send (data); } sock.close (); ctx.term ();
- Connect to ZPWriter and send (socket.write) messages
- Recommend to send multi frames with the first frame as a message id which will be used at the ZPAck
- Create a DEALER or REQ socket
- Set a identity as the above format
- Connect to ZPReader. To sink messages just connect to multiple ZPReader
- Send a "OFFSET" request if you don't have the last offset
- Send a "FETCH" request
# command consists of 1 command frame and variable argument frames
+ ----------+
| command | # command String
+ ----------+
| argument1 | # variable bytes depends on command
+ ----------+
| .... |
+ ----------+
| argumentN |
+ ----------+
# commands
FETCH ( 8_bytes_offset , 8_bytes_max_bytes )
OFFSET ( 8_bytes_last_modified_millis [, 4_bytes_max_entries ] ) # -1 : oldest , -2 : latest
# responses
+ ----------+
| status | # 1 byte common first frame
+ ----------+
- OK : 100
- INVALID_OFFSET : 101
- INVALID_COMMAND : 102
- INTERNAL_ERROR : -1
for fetch request for stream fetch request ( with -1 max bytes )
+ ---------------+ + ---------------+
| Super Frame | | frame 1 |
| + ----------+ | + ---------------+
| | frame1 | | | frame 2 |
| + ----------+ | + ---------------+
| | ... | | | ... |
| + ----------+ | + ---------------+
| | frameN | | | frane N |
| + ----------+ | + ---------------+
+ ---------------+
for offset request
+ ----------+
| offset1 |
+ ----------+
| ... |
+ ----------+
| offsetN |
+ ----------+
Context ctx = ZMQ.context (1); Socket sock = ctx.socket (ZMQ.DEALER); sock.setIdentity (ZPUtils.genTopicIdentity (topic, 0)); // flag = 0 sock.connect ("tcp://127.0.0.1:5556"); // get offset sock.sendMore ("OFFSET"); sock.sendMore (ZPUtils.getBytes (timestamp)); sock.send (ZPUtils.getBytes (entry)); status = ZPUtils.getInt (sock.recv ()); // check status is 100 while (sock.hasReceiveMore ()) { // 8 byte offset long offset = ZPUtils.getLong (sock.recv ()); } // fetch data sock.sendMore ("FETCH"); sock.sendMore (ZPUtils.getBytes (offset)); sock.send (ZPUtils.getBytes (size)); status = ZPUtils.getInt (sock.recv ()); // check status is 100 MsgIterator it = ZPUtils.iterator (sock.recvMsg ()); while (it.hasNext ()) { Msg msg = it.next (); // Do your stuff ... } // store the last offset somewhere lastOffset = offset + it.validBytes () sock.close (); ctx.term ();
// Streaming mode (at 0.1.1) Context ctx = ZMQ.context (1); Socket sock = ctx.socket (ZMQ.DEALER); sock.setIdentity (ZPUtils.genTopicIdentity (topic, 0)); // flag = 0 sock.connect ("tcp://127.0.0.1:5556"); // get offset, refer above // fetch data sock.sendMore ("FETCH"); sock.sendMore (ZPUtils.getBytes (offset)); sock.send (ZPUtils.getBytes (-1L)); // Streaming while (true) { Msg msg = sock.recvMsg (ZMQ.DONTWAIT); if (msg == null) break; // no more message // Do your stuff ... // store the last offset somewhere lastOffset += ZPUtils.validBytes (msg); } sock.close (); ctx.term ();
- Create a DEALER or PULL socket
- Set a identity as the above format ((ZMQ 3.2.2 or above is required for PULL)
- Connect to ZPRelay.
- Doesn't require offset. But you can loose messages if the broker restarted.
Context ctx = ZMQ.context (1); Socket sock = ctx.socket (ZMQ.DEALER); sock.setIdentity (ZPUtils.genTopicIdentity (topic, 0)); // flag = 0 sock.connect ("tcp://127.0.0.1:5558"); while (true) { Msg msg = sock.recvMsg (); if (msg == null) break; // interrupted // Do your stuff ... } sock.close (); ctx.term ();
- Create a SUB socket
- Set subscribe topics which you want to subscribe
- Connect to ZPAck
- ZPAck will send you 3 frames every time ZPWriter flush to disks
# ack consists of 3 frames
+ -----------+
| topic | # topic string
+ -----------+
| message-id | # first frame which was sent to the ZPWriter
+ -----------+
| offset | # 8 bytes offset of the message
+ -----------+
- By connecting to one of ZPController, client can get node and peer information