35、从高级功能拆解4款主流MQ的架构设计与实现

# 创建 queue
channel.queue_declare(queue='route_queue1', 
                      exclusive=True, durable=True)

# 绑定 queue到交换机，并指定 routing key 
channel.queue_bind(exchange='direct_exchange', 
                  queue="route_queue1", routing_key=routingKey)

channel.basic_publish(exchange='direct_exchange',
                      routing_key='A',
                      body=('hello world').encode(),
                     properties=pika.BasicProperties(delivery_mode=2))

Connection connection=null;
Channel channel=null;
try {
   connection = factory.newConnection(); //连接工厂创建连接
   channel = connection.createChannel(); //创建信道
   channel.txSelect(); //开启事务
   channel.queueDeclare(QUEUE_NAME, false, 
            false, false, null);  //绑定队列
   channel.basicPublish("", QUEUE_NAME,
            null, "Hello World!".getBytes(StandardCharsets.UTF_8));
   channel.txCommit();   //提交事务
   System.out.println(" [x] Sent '" + message + "'");
} catch (Exception e){
   e.printStackTrace();
   channel.txRollback(); //回滚事务
}

// 创建了名为 priority_queue 的优先级队列，其最大优先级为 10。 
channel.queue_declare(queue='priority_queue', arguments={'x-max-priority': 10})

// 向优先级队列 priority_queue 发送了一个带有优先级为 5 的消息
channel.basic_publish(exchange='', routing_key='priority_queue', body='Hello World!', properties=pika.BasicProperties(priority=5))

channel.exchange_declare(
    exchange='dlx_direct', exchange_type=ExchangeType.direct)

# 定义死信交换机
channel.queue_declare(queue='dead_queue')
# 死信队列绑定到第一步创建的死信
channel.queue_bind(
    queue='dead_queue', exchange='dlx_direct', routing_key='dead_queue')

channel.queue_declare(
    queue="dlx_queue",
    arguments={
        'x-dead-letter-exchange': 'dlx_direct',
        'x-dead-letter-routing-key': 'dead_queue'
    })

// 启用基于Stomp协议的websocket插件：
rabbitmq-plugins enable rabbitmq_web_stomp

// 启用基于MQTT协议的websocket 插件
rabbitmq-plugins enable rabbitmq_web_mqtt

var ws = new WebSocket('ws://127.0.0.1:15674/ws');
var client = Stomp.over(ws);

 //顺序消息发送。
MessageBuilder messageBuilder = new MessageBuilderImpl();;
Message message = messageBuilder.setTopic("topic")
                //设置消息索引键，可根据关键字精确查找某条消息。
                .setKeys("messageKey")
                //设置消息Tag，用于消费端根据指定Tag过滤消息。
                .setTag("messageTag")
                //设置顺序消息的排序分组，该分组尽量保持离散，避免热点排序分组。
                .setMessageGroup("fifoGroup001")
                //消息体。
                .setBody("messageBody".getBytes())
                .build();
try {
     //发送消息，需要关注发送结果，并捕获失败等异常
     SendReceipt sendReceipt = producer.send(message);
     System.out.println(sendReceipt.getMessageId());
} catch (ClientException e) {
     e.printStackTrace();
}

//定时/延时消息发送
MessageBuilder messageBuilder = new MessageBuilderImpl();;
//以下示例表示：延迟时间为10分钟之后的Unix时间戳。
Long deliverTimeStamp = System.currentTimeMillis() + 10L * 60 * 1000;
Message message = messageBuilder.setTopic("topic")
      //设置消息索引键，可根据关键字精确查找某条消息。
      .setKeys("messageKey")
      //设置消息Tag，用于消费端根据指定Tag过滤消息。
      .setTag("messageTag")
      .setDeliveryTimestamp(deliverTimeStamp)
      //消息体
      .setBody("messageBody".getBytes())
      .build();
try {
    //发送消息，需要关注发送结果，并捕获失败等异常。
    SendReceipt sendReceipt = producer.send(message);
    System.out.println(sendReceipt.getMessageId());
} catch (ClientException e) {
    e.printStackTrace();
}

./bin/mqadmin updatetopic -n localhost:9876 -t TestTopic -c DefaultCluster -a +message.type=TRANSACTION

    //演示demo，模拟订单表查询服务，用来确认订单事务是否提交成功。
    private static boolean checkOrderById(String orderId) {
        return true;
    }
    //演示demo，模拟本地事务的执行结果。
    private static boolean doLocalTransaction() {
        return true;
    }
    public static void main(String[] args) throws ClientException {
        ClientServiceProvider provider = new ClientServiceProvider();
        MessageBuilder messageBuilder = new MessageBuilderImpl();
        //构造事务生产者：事务消息需要生产者构建一个事务检查器，用于检查确认异常半事务的中间状态。
        Producer producer = provider.newProducerBuilder()
                .setTransactionChecker(messageView -> {
                    /**
                     * 事务检查器一般是根据业务的ID去检查本地事务是否正确提交还是回滚，此处以订单ID属性为例。
                     * 在订单表找到了这个订单，说明本地事务插入订单的操作已经正确提交；如果订单表没有订单，说明本地事务已经回滚。
                     */
                    final String orderId = messageView.getProperties().get("OrderId");
                    if (Strings.isNullOrEmpty(orderId)) {
                        // 错误的消息，直接返回Rollback。
                        return TransactionResolution.ROLLBACK;
                    }
                    return checkOrderById(orderId) ? TransactionResolution.COMMIT : TransactionResolution.ROLLBACK;
                })
                .build();
        //开启事务分支。
        final Transaction transaction;
        try {
            transaction = producer.beginTransaction();
        } catch (ClientException e) {
            e.printStackTrace();
            //事务分支开启失败，直接退出。
            return;
        }
        Message message = messageBuilder.setTopic("topic")
                //设置消息索引键，可根据关键字精确查找某条消息。
                .setKeys("messageKey")
                //设置消息Tag，用于消费端根据指定Tag过滤消息。
                .setTag("messageTag")
                //一般事务消息都会设置一个本地事务关联的唯一ID，用来做本地事务回查的校验。
                .addProperty("OrderId", "xxx")
                //消息体。
                .setBody("messageBody".getBytes())
                .build();
        //发送半事务消息
        final SendReceipt sendReceipt;
        try {
            sendReceipt = producer.send(message, transaction);
        } catch (ClientException e) {
            //半事务消息发送失败，事务可以直接退出并回滚。
            return;
        }
        /**
         * 执行本地事务，并确定本地事务结果。
         * 1. 如果本地事务提交成功，则提交消息事务。
         * 2. 如果本地事务提交失败，则回滚消息事务。
         * 3. 如果本地事务未知异常，则不处理，等待事务消息回查。
         *
         */
        boolean localTransactionOk = doLocalTransaction();
        if (localTransactionOk) {
            try {
                transaction.commit();
            } catch (ClientException e) {
                // 业务可以自身对实时性的要求选择是否重试，如果放弃重试，可以依赖事务消息回查机制进行事务状态的提交。
                e.printStackTrace();
            }
        } else {
            try {
                transaction.rollback();
            } catch (ClientException e) {
                // 建议记录异常信息，回滚异常时可以无需重试，依赖事务消息回查机制进行事务状态的提交。
                e.printStackTrace();
            }
        }
    }

 // 1. 创建DefaultMQPushConsumer实例
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("DLQ_CONSUMER");
// 2. 设置NameServer地址
consumer.setNamesrvAddr("127.0.0.1:9876");
// 3. 设置死信队列名
consumer.setDLQName("DLQ_NAME");
// 4. 设置处理死信队列消息的消费者组
consumer.setDLQConsumerGroup("DLQ_CONSUMER_GROUP");
// 5. 启动消费者实例，连接NameServer
consumer.start();
}

// 设置偏移量
long offset = 10;
 while (true) {
    // 拉取消息
    PullResult pullResult =consumer.pull("queue1", "*", offset, 32);
    System.out.println(pullResult);

    // 更新偏移量
    offset = pullResult.getNextBeginOffset();

    // 消费消息并设置延迟，模拟业务处理
    Thread.sleep(1000);
} 

// 设置查询消息的时间戳（毫秒）
long timestamp = System.currentTimeMillis() - (1000 * 60 * 60);

// 获取与时间戳最近的消息偏移量
long offset = consumer.searchOffset(mq, timestamp);

while (true) {
    // 拉取消息
    PullResult pullResult = consumer.pull(mq, "*", offset, 32);
    System.out.println(pullResult);
    // 更新偏移量
    offset = pullResult.getNextBeginOffset();

   // 消费消息并设置延迟，模拟业务处理
    Thread.sleep(1000);
}

// 创建 DefaultMQAdminExt 对象
DefaultMQAdminExt mqAdmin = new DefaultMQAdminExt();
// 设置 NameServer 地址
mqAdmin.setNamesrvAddr("localhost:9876");
// 启动
mqAdmin.start();

// 查询消息 ID
String msgId = "k1";

// 根据消息 ID 查询消息
MessageExt message = mqAdmin.viewMessage("TopicTest", msgId);

// 输出消息内容
if (message != null) {
    System.out.println("Message: " + message);
} else {
    System.out.println("Message not found.");
}

Properties props = new Properties();
props.put(ProducerConfig.LINGER_MS_CONFIG, "1000");

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

producer.send(new ProducerRecord<>(topic, 
    "key1",""code:1,message:" + Time.SYSTEM.nanoseconds()));

Properties props = new Properties();
props.put("bootstrap.servers", bootstrap);
props.put("retries", 2); // 重试次数
props.put("batch.size", 100); // 批量发送大小
props.put("buffer.memory", 33554432); // 缓存大小，根据本机内存大小配置
props.put("linger.ms", 1000); // 发送频率，满足任务一个条件发送
props.put("client.id", clientId); // 发送端id,便于统计 "token#sfdiewrnxkcvvulsdfsdfdsijuiewrewr"
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("enable.idempotence", true); // 设置幂等性
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
Long startTime = Time.SYSTEM.milliseconds();
Integer count = 0;
while (true) {
    try {
        // 开启事务
        // 发送消息到producer-syn
        producer.send(new ProducerRecord<>(topic, "msg1");
       
    } catch (Exception e) {
        e.printStackTrace();
    }
}

Properties props = new Properties();
props.put("bootstrap.servers", bootstrap);
props.put("retries", 2); // 重试次数
props.put("batch.size", 100); // 批量发送大小
props.put("buffer.memory", 33554432); // 缓存大小，根据本机内存大小配置
props.put("linger.ms", 1000); // 发送频率，满足任务一个条件发送
props.put("client.id", "producer-txn-test"); // 发送端id,便于统计
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("transactional.id", txnId); // 每台机器唯一
props.put("enable.idempotence", true); // 设置幂等性
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();
Long startTime = Time.SYSTEM.milliseconds();
Integer count = 0;
while (true) {
    try {
        // 开启事务
        producer.beginTransaction();
        // 发送消息到producer-syn
        producer.send(new ProducerRecord<>(topic, "message"));
    } catch (Exception e) {
        e.printStackTrace();
        // 终止事务
        producer.abortTransaction();
    }
}

timestamp: 1693001346933 offset: 62369391
timestamp: 1693001346957 offset: 62369395
timestamp: 1693001347033 offset: 62369397
timestamp: 1693001420165 offset: 62369402
timestamp: 1693001420203 offset: 62369408

offset: 62369391 position: 4462
offset: 62369395 position: 9664
offset: 62369397 position: 13986
offset: 62369402 position: 18309
offset: 62369408 position: 23699
offset: 62369414 position: 29882
offset: 62369418 position: 35910