1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

// ====== 生产者示例 ======
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");              // 强一致性
props.put("retries", 3);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

Producer<String, String> producer = new KafkaProducer<>(props);

// 发送消息
producer.send(new ProducerRecord<>("task-status", 
    "job-001", 
    "{\"status\":\"RUNNING\",\"timestamp\":\"2025-04-20\"}"));
producer.close();

// ====== 消费者示例 ======
Properties consumerProps = new Properties();
consumerProps.put("bootstrap.servers", "localhost:9092");
consumerProps.put("group.id", "task-processor");
consumerProps.put("enable.auto.commit", "true");
consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

Consumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(Arrays.asList("task-status"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.println("Received: " + record.value());
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

// ====== 生产者示例 ======
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setUsername("guest");
factory.setPassword("guest");

try (Connection connection = factory.newConnection();
     Channel channel = connection.createChannel()) {

    // 声明 Exchange
    channel.exchangeDeclare("task-exchange", BuiltinExchangeType.DIRECT, true);

    // 声明 Queue
    channel.queueDeclare("task-queue", true, false, false, null);

    // 绑定
    channel.queueBind("task-queue", "task-exchange", "task.routing.key");

    // 发布消息
    String message = "{\"jobId\":\"job-001\",\"status\":\"RUNNING\"}";
    channel.basicPublish("task-exchange", "task.routing.key",
        MessageProperties.PERSISTENT_TEXT_PLAIN,
        message.getBytes(StandardCharsets.UTF_8));
}

// ====== 消费者示例 ======
try (Connection connection = factory.newConnection();
     Channel channel = connection.createChannel()) {

    channel.queueDeclare("task-queue", true, false, false, null);

    DeliverCallback deliverCallback = (consumerTag, delivery) -> {
        String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
        System.out.println("Received: " + message);

        // 手动确认
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    };

    channel.basicConsume("task-queue", false, deliverCallback, consumerTag -> {});
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

# 内存相关
vm_memory_high_watermark.relative = 0.4       # 使用40%内存
vm_memory_high_watermark_paging_ratio = 0.5   # 达到50%时开始换页

# 磁盘相关
disk_free_limit.relative = 1.0                 # 保留1倍内存大小的磁盘空间
disk_free_limit.absolute = 2GB                 # 最小2GB

# 网络相关
listeners.tcp.default = 5672
num_acceptors.tcp = 10                          # 接受器线程数

# 队列镜像
queue.master_locator = min-masters              # 主队列位置策略

# 消息持久化
message_store.ignore_confirms = false           # 不忽略确认

# 集群配置
cluster_partition_handling = pause_minority     # 分区处理策略

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# Quorum Queue (推荐)
quorum_queue.initial_group_size = 3           # 初始副本数

# Classic Queue 镜像 (旧方式)
policy definition:
{
  "ha-mode": "exactly",
  "ha-params": 2,
  "ha-sync-mode": "automatic"
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

// ====== 生产者示例 ======
DefaultMQProducer producer = new DefaultMQProducer("task-producer-group");
producer.setNamesrvAddr("localhost:9876");
producer.start();

// 普通消息
Message msg = new Message("task-status",
    "TAG_RUNNING",
    "job-001",
    "{\"jobId\":\"job-001\",\"status\":\"RUNNING\"}".getBytes());
SendResult result = producer.send(msg);

// 延迟消息 (级别 3 = 10秒)
msg.setDelayTimeLevel(3);
producer.send(msg);

// 事务消息
TransactionMQProducer txProducer = new TransactionMQProducer("tx-producer-group");
txProducer.setTransactionListener(new TransactionListener() {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 执行本地事务
        return LocalTransactionState.COMMIT_MESSAGE;
    }
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 事务回查
        return LocalTransactionState.COMMIT_MESSAGE;
    }
});
txProducer.sendMessageInTransaction(msg, null);

// ====== 消费者示例 ======
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("task-consumer-group");
consumer.setNamesrvAddr("localhost:9876");
consumer.subscribe("task-status", "TAG_RUNNING || TAG_SUCCESS");
consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                   ConsumeConcurrentlyContext context) {
        for (MessageExt msg : msgs) {
            System.out.println("Received: " + new String(msg.getBody()));
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});
consumer.start();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

// ====== Go 语言示例 ======
import "github.com/nats-io/nats.go"

// 连接
nc, _ := nats.Connect(nats.DefaultURL)

// ====== 发布者 ======
nc.Publish("task.running", []byte(`{"jobId":"job-001"}`))

// 请求-响应模式
msg, _ := nc.Request("service.hello", []byte("hello"), time.Second)

// ====== 订阅者 ======
nc.Subscribe("task.running", func(m *nats.Msg) {
    fmt.Printf("Received: %s\n", string(m.Data))
})

// 队列组（负载均衡）
nc.QueueSubscribe("task.running", "worker-group", func(m *nats.Msg) {
    fmt.Printf("Worker received: %s\n", string(m.Data))
})

// ====== JetStream (持久化) ======
js, _ := nc.JetStream()

// 创建流
js.AddStream(&nats.StreamConfig{
    Name:     "TASKS",
    Subjects: []string{"task.>"},
    Storage:  nats.FileStorage,
})

// 发布持久化消息
js.Publish("task.running", []byte(`{"jobId":"job-001"}`))

// 订阅并确认
sub, _ := js.Subscribe("task.running", func(m *nats.Msg) {
    fmt.Printf("Received: %s\n", string(m.Data))
    m.Ack()  // 确认消息
})

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

# 监听端口
port: 4222
http: 8222          # 监控端点
websocket: 8080

# 集群配置
cluster {
  name: nats-cluster
  listen: 0.0.0.0:6222
  routes: [
    nats-route://nats-1:6222,
    nats-route://nats-2:6222,
    nats-route://nats-3:6222
  ]
}

# 日志
debug: false
trace: false
logtime: true

# 连接限制
max_connections: 65536
max_subscriptions: 0  # 0 = 无限制

# 认证
authorization {
  user: nats
  password: secret
  timeout: 1
}

# JetStream 配置 (如果启用持久化)
jetstream {
  store_dir: /data/jetstream
  max_memory_store: 1GB
  max_file_store: 10GB
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# Stream 配置示例
name: TASKS
subjects: ["task.>"]
retention: limits    # limits / workqueue / interest
max_consumers: -1    # 无限制
max_msgs: -1         # 无限制
max_bytes: -1        # 无限制
max_age: 168h        # 7天
storage: file        # file / memory
discard: old         # 旧消息先丢弃
num_replicas: 3      # 副本数
duplicate_window: 2m # 去重窗口

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

// ====== Java 客户端示例 ======
PulsarClient client = PulsarClient.builder()
    .serviceUrl("pulsar://localhost:6650")
    .build();

// ====== 生产者 ======
Producer<String> producer = client.newProducer(Schema.STRING)
    .topic("persistent://my-tenant/my-namespace/task-status")
    .create();

MessageId msgId = producer.newMessage()
    .key("job-001")
    .value("{\"status\":\"RUNNING\"}")
    .send();

// ====== 消费者 ======
Consumer<String> consumer = client.newConsumer(Schema.STRING)
    .topic("persistent://my-tenant/my-namespace/task-status")
    .subscriptionName("task-processor")
    .subscriptionType(SubscriptionType.Shared)  // 负载均衡
    .subscribe();

while (true) {
    Message<String> msg = consumer.receive();
    try {
        System.out.println("Received: " + msg.getValue());
        consumer.acknowledge(msg);  // 确认
    } catch (Exception e) {
        consumer.negativeAcknowledge(msg);  // 重试
    }
}

// ====== 地理复制 (Replication) ======
// 创建跨地域复制的 Topic
producer = client.newProducer(Schema.STRING)
    .topic("persistent://my-tenant/my-namespace/global-topic")
    .create();

// 消息自动复制到配置的集群
producer.send("Replicated message");

// ====== 分层存储配置 ======
// 旧数据自动卸载到 S3 (在 broker 配置中)
// managedLedgerOffloadThresholdInSeconds=86400  // 1天后卸载
// 消费者透明访问，无需关心数据在哪里

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

import redis

# 发布
r = redis.Redis()
r.publish('task-notifications', '{"jobId":"001","status":"RUNNING"}')

# 订阅
pubsub = r.pubsub()
pubsub.subscribe('task-notifications')
for message in pubsub.listen():
    if message['type'] == 'message':
        print(message['data'])

1
2
3
4
5
6

# 添加消息
r.xadd('task-stream', {'jobId': '001', 'status': 'RUNNING'})

# 消费组消费
r.xgroup_create('task-stream', 'worker-group', id='0', mkstream=True)
messages = r.xreadgroup('worker-group', 'consumer-1', {'task-stream': '>'}, count=10)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

-- 发送通知
NOTIFY task_channel, '{"jobId":"001","status":"RUNNING"}';

-- 或使用触发器自动发送
CREATE OR REPLACE FUNCTION notify_task_change()
RETURNS trigger AS $$
BEGIN
    PERFORM pg_notify('task_channel', row_to_json(NEW)::text);
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER task_status_trigger
    AFTER UPDATE ON tasks
    FOR EACH ROW EXECUTE FUNCTION notify_task_change();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

CREATE TABLE task_queue (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    payload JSON NOT NULL,
    status ENUM('pending', 'processing', 'done', 'failed') DEFAULT 'pending',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    locked_at TIMESTAMP NULL,
    INDEX idx_status_created (status, created_at)
);

-- 领取任务 (使用 SELECT ... FOR UPDATE SKIP LOCKED)
START TRANSACTION;
SELECT * FROM task_queue 
WHERE status = 'pending' 
ORDER BY created_at 
LIMIT 1 FOR UPDATE SKIP LOCKED;
-- 更新状态为 processing...
COMMIT;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

// JMS API 使用
ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
Connection connection = factory.createConnection();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Queue queue = session.createQueue("task-queue");

// 发送
MessageProducer producer = session.createProducer(queue);
TextMessage message = session.createTextMessage("{\"jobId\":\"001\"}");
producer.send(message);

// 接收
MessageConsumer consumer = session.createConsumer(queue);
TextMessage received = (TextMessage) consumer.receive();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

import boto3

sns = boto3.client('sns', region_name='us-east-1')

# 创建 Topic
topic_arn = sns.create_topic(Name='task-status')['TopicArn']

# 发布消息
sns.publish(
    TopicArn=topic_arn,
    Message='{"jobId":"001","status":"RUNNING"}',
    MessageAttributes={
        'jobId': {'DataType': 'String', 'StringValue': '001'}
    }
)

# 订阅 - SQS
sns.subscribe(
    TopicArn=topic_arn,
    Protocol='sqs',
    Endpoint=queue_arn
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

sqs = boto3.client('sqs', region_name='us-east-1')

# 创建队列
queue_url = sqs.create_queue(
    QueueName='task-queue.fifo',  # .fifo 后缀表示 FIFO 队列
    Attributes={
        'FifoQueue': 'true',
        'ContentBasedDeduplication': 'true',
        'VisibilityTimeout': '300',  # 5分钟可见性超时
        'MessageRetentionPeriod': '86400'  # 保留1天
    }
)['QueueUrl']

# 发送消息
sqs.send_message(
    QueueUrl=queue_url,
    MessageBody='{"jobId":"001","status":"RUNNING"}',
    MessageGroupId='job-group',  # FIFO 需要
    MessageDeduplicationId='001'
)

# 接收消息
response = sqs.receive_message(
    QueueUrl=queue_url,
    MaxNumberOfMessages=10,
    WaitTimeSeconds=20  # 长轮询
)

for msg in response.get('Messages', []):
    print(msg['Body'])
    # 删除消息
    sqs.delete_message(
        QueueUrl=queue_url,
        ReceiptHandle=msg['ReceiptHandle']
    )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

import com.azure.messaging.servicebus.*;

// 客户端
ServiceBusProcessorClient processor = new ServiceBusClientBuilder()
    .connectionString("Endpoint=sb://xxx.servicebus.windows.net/;...")
    .processor()
    .queueName("task-queue")
    .processMessage(context -> {
        System.out.println("Received: " + context.getMessage().getBody().toString());
        context.complete();  // 确认消息
    })
    .processError(context -> {
        System.out.println("Error: " + context.getException());
    })
    .buildProcessorClient();

processor.start();

// 发送消息
ServiceBusSenderClient sender = new ServiceBusClientBuilder()
    .connectionString(connectionString)
    .sender()
    .queueName("task-queue")
    .buildSenderClient();

sender.sendMessage(new ServiceBusMessage("{\"jobId\":\"001\"}"));

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

from google.cloud import pubsub_v1
import json

# ====== 生产者 ======
publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path("your-project-id", "task-topic")

# 发布消息
message = {"jobId": "001", "status": "RUNNING"}
future = publisher.publish(
    topic_path,
    json.dumps(message).encode("utf-8"),
    jobId="001",  # 属性
    orderingKey="job-001"  # 排序键，保证有序
)
message_id = future.result()
print(f"Published message ID: {message_id}")

# ====== 消费者 (Push) ======
# 使用 Cloud Functions 或 Cloud Run 自动触发

# ====== 消费者 (Pull) ======
subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(
    "your-project-id", "task-subscription"
)

def callback(message):
    print(f"Received: {message.data.decode()}")
    message.ack()  # 确认

streaming_pull_future = subscriber.subscribe(
    subscription_path, 
    callback=callback
)

print(f"Listening for messages on {subscription_path}")
try:
    streaming_pull_future.result()
except KeyboardInterrupt:
    streaming_pull_future.cancel()

# ====== 配置 Exactly-once ======
# 创建支持精确一次的订阅
subscription = subscriber.create_subscription(
    request={
        "name": subscription_path,
        "topic": topic_path,
        "enable_exactly_once_delivery": True,
        "ack_deadline_seconds": 60,
        "message_retention_duration": {"seconds": 604800}  # 7天
    }
)

# ====== 死信队列 ======
# 创建带死信队列的订阅
subscription = subscriber.create_subscription(
    request={
        "name": subscription_path,
        "topic": topic_path,
        "dead_letter_policy": {
            "dead_letter_topic": dead_letter_topic_path,
            "max_delivery_attempts": 5
        }
    }
)