ZeroMQ

通知類型

# bitcoin.conf

# 區塊雜湊通知
zmqpubhashblock=tcp://127.0.0.1:28332

# 交易雜湊通知
zmqpubhashtx=tcp://127.0.0.1:28333

# 原始區塊通知
zmqpubrawblock=tcp://127.0.0.1:28334

# 原始交易通知
zmqpubrawtx=tcp://127.0.0.1:28335

# 序列通知
zmqpubsequence=tcp://127.0.0.1:28336

# 高水位標記（消息隊列大小）
zmqpubhashblockhwm=1000
zmqpubhashtxhwm=1000

import zmq
import binascii

# 創建 ZMQ 上下文
context = zmq.Context()
socket = context.socket(zmq.SUB)

# 連接到 Bitcoin Core
socket.connect("tcp://127.0.0.1:28332")

# 訂閱所有消息（空字串 = 訂閱所有）
socket.setsockopt_string(zmq.SUBSCRIBE, "hashblock")

print("等待區塊通知...")

while True:
    # 接收多部分消息
    topic = socket.recv()
    body = socket.recv()
    sequence = socket.recv()

    # 解析區塊雜湊（小端序）
    block_hash = binascii.hexlify(body[::-1]).decode()
    seq_num = int.from_bytes(sequence, 'little')

    print(f"新區塊: {block_hash}")
    print(f"序列號: {seq_num}")

import * as zmq from 'zeromq';

async function subscribeToBlocks() {
  const socket = new zmq.Subscriber();

  socket.connect('tcp://127.0.0.1:28332');
  socket.subscribe('hashblock');

  console.log('等待區塊通知...');

  for await (const [topic, body, sequence] of socket) {
    // 反轉字節順序（小端 -> 大端）
    const blockHash = Buffer.from(body)
      .reverse()
      .toString('hex');

    const seqNum = sequence.readUInt32LE(0);

    console.log(`新區塊: ${blockHash}`);
    console.log(`序列號: ${seqNum}`);
  }
}

subscribeToBlocks();

use zmq;

fn main() {
    let context = zmq::Context::new();
    let socket = context.socket(zmq::SUB).unwrap();

    socket.connect("tcp://127.0.0.1:28332").unwrap();
    socket.set_subscribe(b"hashblock").unwrap();

    println!("等待區塊通知...");

    loop {
        let topic = socket.recv_bytes(0).unwrap();
        let body = socket.recv_bytes(0).unwrap();
        let sequence = socket.recv_bytes(0).unwrap();

        // 反轉字節順序
        let mut hash = body.clone();
        hash.reverse();
        let block_hash = hex::encode(&hash);

        let seq_num = u32::from_le_bytes(
            sequence[..4].try_into().unwrap()
        );

        println!("新區塊: {}", block_hash);
        println!("序列號: {}", seq_num);
    }
}

序列消息格式：

| 字段 | 長度 | 描述 |
|------|------|------|
| hash | 32   | 區塊/交易雜湊 |
| label| 1    | 事件類型 |
| sequence | 4 | 序列號 |

事件類型（label）：
'A' = 交易加入 mempool
'R' = 交易從 mempool 移除
'C' = 區塊連接
'D' = 區塊斷開（重組）

# 訂閱序列通知
socket.setsockopt_string(zmq.SUBSCRIBE, "sequence")

while True:
    topic = socket.recv()
    body = socket.recv()

    hash_bytes = body[:32]
    label = chr(body[32])
    sequence = int.from_bytes(body[33:37], 'little')

    hash_hex = binascii.hexlify(hash_bytes[::-1]).decode()

    events = {
        'A': '交易加入 mempool',
        'R': '交易移除 mempool',
        'C': '區塊連接',
        'D': '區塊斷開'
    }

    print(f"{events.get(label, '未知')}: {hash_hex[:16]}...")

# 使用序列號檢測丟失的消息
last_sequence = None

while True:
    topic = socket.recv()
    body = socket.recv()
    sequence = socket.recv()

    seq_num = int.from_bytes(sequence, 'little')

    if last_sequence is not None:
        if seq_num != last_sequence + 1:
            print(f"警告：可能丟失消息 {last_sequence + 1} - {seq_num - 1}")
            # 需要通過 RPC 重新同步

    last_sequence = seq_num

    # 處理消息...

# 檢查 ZMQ 配置
bitcoin-cli getzmqnotifications

# 輸出示例
[
  {
    "type": "pubhashblock",
    "address": "tcp://127.0.0.1:28332",
    "hwm": 1000
  },
  {
    "type": "pubhashtx",
    "address": "tcp://127.0.0.1:28333",
    "hwm": 1000
  }
]

# 使用 Python 快速測試
python3 -c "
import zmq
ctx = zmq.Context()
s = ctx.socket(zmq.SUB)
s.connect('tcp://127.0.0.1:28332')
s.subscribe('')
print('收到:', s.recv(), s.recv())
"