第5章：RocketMQ高级特性
5.1 顺序消息
5.1.1 顺序消息概念
from enum import Enum
from typing import List, Dict, Optional, Callable
from dataclasses import dataclass
from abc import ABC, abstractmethod
import time
import threading
import hashlib
# 顺序类型枚举
class OrderType(Enum):
GLOBAL_ORDER = "GLOBAL_ORDER" # 全局顺序
PARTITION_ORDER = "PARTITION_ORDER" # 分区顺序
# 顺序消息状态
class OrderMessageStatus(Enum):
SUCCESS = "SUCCESS"
ROLLBACK = "ROLLBACK"
COMMIT = "COMMIT"
UNKNOWN = "UNKNOWN"
# 顺序消息
@dataclass
class OrderMessage:
topic: str
tags: str
keys: str
body: bytes
order_id: str # 顺序标识
order_type: OrderType = OrderType.PARTITION_ORDER
properties: Dict[str, str] = None
def __post_init__(self):
if self.properties is None:
self.properties = {}
# 设置顺序标识属性
self.properties["ORDER_ID"] = self.order_id
self.properties["ORDER_TYPE"] = self.order_type.value
def get_order_key(self) -> str:
"""获取顺序键"""
return self.order_id
# 顺序消息选择器
class OrderMessageQueueSelector:
"""顺序消息队列选择器"""
def select(self, queues: List['MessageQueue'], message: OrderMessage, arg: str) -> 'MessageQueue':
"""选择消息队列"""
if not queues:
raise ValueError("消息队列列表为空")
# 根据顺序ID计算队列索引
order_key = message.get_order_key()
hash_value = self._hash(order_key)
index = hash_value % len(queues)
return queues[index]
def _hash(self, key: str) -> int:
"""计算哈希值"""
return hash(key) & 0x7fffffff
# 顺序消息生产者
class OrderMessageProducer:
"""顺序消息生产者"""
def __init__(self, producer_config: 'ProducerConfig'):
self.config = producer_config
self.producer = None # 底层生产者
self.queue_selector = OrderMessageQueueSelector()
self.send_stats = {
"total_sent": 0,
"success_sent": 0,
"failed_sent": 0
}
def start(self):
"""启动顺序消息生产者"""
# 初始化底层生产者
from rocketmq_producer import DefaultMQProducer
self.producer = DefaultMQProducer(self.config)
self.producer.start()
print("顺序消息生产者启动成功")
def shutdown(self):
"""关闭顺序消息生产者"""
if self.producer:
self.producer.shutdown()
print("顺序消息生产者关闭成功")
def send_order_message(self, message: OrderMessage) -> 'SendResult':
"""发送顺序消息"""
try:
# 转换为普通消息
normal_message = self._convert_to_message(message)
# 使用顺序选择器发送
result = self.producer.send(
message=normal_message,
selector=self.queue_selector,
arg=message.order_id
)
self.send_stats["total_sent"] += 1
self.send_stats["success_sent"] += 1
return result
except Exception as e:
self.send_stats["total_sent"] += 1
self.send_stats["failed_sent"] += 1
raise Exception(f"发送顺序消息失败: {e}")
def send_order_message_async(self, message: OrderMessage,
callback: 'SendCallback'):
"""异步发送顺序消息"""
try:
normal_message = self._convert_to_message(message)
# 包装回调
wrapped_callback = OrderSendCallback(callback, self.send_stats)
self.producer.send_async(
message=normal_message,
selector=self.queue_selector,
arg=message.order_id,
callback=wrapped_callback
)
except Exception as e:
self.send_stats["total_sent"] += 1
self.send_stats["failed_sent"] += 1
callback.on_exception(e)
def _convert_to_message(self, order_message: OrderMessage) -> 'Message':
"""转换为普通消息"""
from rocketmq_producer import Message
message = Message(
topic=order_message.topic,
tags=order_message.tags,
keys=order_message.keys,
body=order_message.body
)
# 复制属性
for key, value in order_message.properties.items():
message.put_property(key, value)
return message
def get_send_stats(self) -> Dict[str, int]:
"""获取发送统计"""
return self.send_stats.copy()
# 顺序发送回调包装器
class OrderSendCallback:
"""顺序发送回调包装器"""
def __init__(self, callback: 'SendCallback', stats: Dict[str, int]):
self.callback = callback
self.stats = stats
def on_success(self, result: 'SendResult'):
"""发送成功回调"""
self.stats["total_sent"] += 1
self.stats["success_sent"] += 1
self.callback.on_success(result)
def on_exception(self, exception: Exception):
"""发送异常回调"""
self.stats["total_sent"] += 1
self.stats["failed_sent"] += 1
self.callback.on_exception(exception)
5.1.2 顺序消息消费
# 顺序消息消费者
class OrderMessageConsumer:
"""顺序消息消费者"""
def __init__(self, consumer_config: 'ConsumerConfig'):
self.config = consumer_config
# 强制设置为顺序消费模式
self.config.consume_mode = ConsumeMode.ORDERLY
self.consumer = None
self.order_listener: Optional['OrderMessageListener'] = None
self.consume_stats = {
"total_consumed": 0,
"success_consumed": 0,
"failed_consumed": 0
}
def start(self):
"""启动顺序消息消费者"""
if not self.order_listener:
raise Exception("未设置顺序消息监听器")
# 初始化底层消费者
from rocketmq_consumer import DefaultMQPushConsumer
self.consumer = DefaultMQPushConsumer(self.config)
# 注册顺序消息监听器
wrapped_listener = OrderMessageListenerWrapper(
self.order_listener, self.consume_stats
)
self.consumer.register_message_listener(wrapped_listener)
self.consumer.start()
print("顺序消息消费者启动成功")
def shutdown(self):
"""关闭顺序消息消费者"""
if self.consumer:
self.consumer.shutdown()
print("顺序消息消费者关闭成功")
def subscribe(self, topic: str, sub_expression: str = "*"):
"""订阅主题"""
if self.consumer:
self.consumer.subscribe(topic, sub_expression)
def register_order_listener(self, listener: 'OrderMessageListener'):
"""注册顺序消息监听器"""
self.order_listener = listener
def get_consume_stats(self) -> Dict[str, int]:
"""获取消费统计"""
return self.consume_stats.copy()
# 顺序消息监听器接口
class OrderMessageListener(ABC):
"""顺序消息监听器接口"""
@abstractmethod
def consume_order_message(self, messages: List['MessageExt'],
context: 'ConsumeOrderlyContext') -> 'ConsumeOrderlyResult':
"""消费顺序消息"""
pass
# 顺序消费上下文
@dataclass
class ConsumeOrderlyContext:
consumer_group: str
message_queue: 'MessageQueue'
auto_commit: bool = True
suspend_current_queue_time_millis: int = 1000
def set_auto_commit(self, auto_commit: bool):
"""设置自动提交"""
self.auto_commit = auto_commit
def set_suspend_current_queue_time_millis(self, time_millis: int):
"""设置暂停当前队列时间"""
self.suspend_current_queue_time_millis = time_millis
# 顺序消费结果
class ConsumeOrderlyResult(Enum):
SUCCESS = "SUCCESS"
SUSPEND_CURRENT_QUEUE_A_MOMENT = "SUSPEND_CURRENT_QUEUE_A_MOMENT"
# 顺序消息监听器包装器
class OrderMessageListenerWrapper:
"""顺序消息监听器包装器"""
def __init__(self, order_listener: OrderMessageListener,
stats: Dict[str, int]):
self.order_listener = order_listener
self.stats = stats
def consume_message(self, messages: List['MessageExt'],
context: 'ConsumeContext') -> 'ConsumeResult':
"""消费消息（适配器方法）"""
try:
# 转换上下文
order_context = ConsumeOrderlyContext(
consumer_group=context.consumer_group,
message_queue=context.message_queue
)
# 调用顺序监听器
result = self.order_listener.consume_order_message(messages, order_context)
# 更新统计
if result == ConsumeOrderlyResult.SUCCESS:
self.stats["total_consumed"] += len(messages)
self.stats["success_consumed"] += len(messages)
return ConsumeResult.SUCCESS
else:
self.stats["total_consumed"] += len(messages)
self.stats["failed_consumed"] += len(messages)
return ConsumeResult.RECONSUME_LATER
except Exception as e:
print(f"顺序消费异常: {e}")
self.stats["total_consumed"] += len(messages)
self.stats["failed_consumed"] += len(messages)
return ConsumeResult.RECONSUME_LATER
# 业务顺序消息监听器
class BusinessOrderMessageListener(OrderMessageListener):
"""业务顺序消息监听器"""
def __init__(self, business_handler: Callable[[List['MessageExt']], bool]):
self.business_handler = business_handler
self.processed_orders = {} # 已处理的订单
self.lock = threading.Lock()
def consume_order_message(self, messages: List['MessageExt'],
context: ConsumeOrderlyContext) -> ConsumeOrderlyResult:
"""消费顺序消息"""
try:
# 检查消息顺序
if not self._check_message_order(messages):
print("消息顺序检查失败，暂停队列")
context.set_suspend_current_queue_time_millis(5000)
return ConsumeOrderlyResult.SUSPEND_CURRENT_QUEUE_A_MOMENT
# 处理业务逻辑
success = self.business_handler(messages)
if success:
# 记录已处理的订单
self._record_processed_orders(messages)
return ConsumeOrderlyResult.SUCCESS
else:
print("业务处理失败，暂停队列")
return ConsumeOrderlyResult.SUSPEND_CURRENT_QUEUE_A_MOMENT
except Exception as e:
print(f"顺序消费异常: {e}")
return ConsumeOrderlyResult.SUSPEND_CURRENT_QUEUE_A_MOMENT
def _check_message_order(self, messages: List['MessageExt']) -> bool:
"""检查消息顺序"""
for msg in messages:
order_id = msg.get_property("ORDER_ID")
if not order_id:
continue
# 检查是否按顺序到达
with self.lock:
last_offset = self.processed_orders.get(order_id, -1)
if msg.queue_offset <= last_offset:
return False
return True
def _record_processed_orders(self, messages: List['MessageExt']):
"""记录已处理的订单"""
with self.lock:
for msg in messages:
order_id = msg.get_property("ORDER_ID")
if order_id:
self.processed_orders[order_id] = msg.queue_offset
5.2 事务消息
5.2.1 事务消息概念
# 事务消息状态
class LocalTransactionState(Enum):
COMMIT_MESSAGE = "COMMIT_MESSAGE" # 提交事务
ROLLBACK_MESSAGE = "ROLLBACK_MESSAGE" # 回滚事务
UNKNOWN = "UNKNOWN" # 未知状态
# 事务消息
@dataclass
class TransactionMessage:
topic: str
tags: str
keys: str
body: bytes
transaction_id: str # 事务ID
properties: Dict[str, str] = None
def __post_init__(self):
if self.properties is None:
self.properties = {}
# 设置事务标识
self.properties["TRANSACTION_ID"] = self.transaction_id
self.properties["TRAN_MSG"] = "true"
# 事务监听器接口
class TransactionListener(ABC):
"""事务监听器接口"""
@abstractmethod
def execute_local_transaction(self, message: TransactionMessage,
arg: object) -> LocalTransactionState:
"""执行本地事务"""
pass
@abstractmethod
def check_local_transaction(self, message: TransactionMessage) -> LocalTransactionState:
"""检查本地事务状态"""
pass
# 事务消息生产者
class TransactionMQProducer:
"""事务消息生产者"""
def __init__(self, producer_config: 'ProducerConfig'):
self.config = producer_config
self.producer = None
self.transaction_listener: Optional[TransactionListener] = None
self.transaction_check_executor = None
self.transaction_states = {} # 事务状态缓存
self.transaction_stats = {
"total_transactions": 0,
"committed_transactions": 0,
"rollback_transactions": 0,
"unknown_transactions": 0
}
self.lock = threading.Lock()
def start(self):
"""启动事务消息生产者"""
if not self.transaction_listener:
raise Exception("未设置事务监听器")
# 初始化底层生产者
from rocketmq_producer import DefaultMQProducer
self.producer = DefaultMQProducer(self.config)
# 启动事务检查线程池
from concurrent.futures import ThreadPoolExecutor
self.transaction_check_executor = ThreadPoolExecutor(
max_workers=5,
thread_name_prefix="TransactionCheck"
)
self.producer.start()
print("事务消息生产者启动成功")
def shutdown(self):
"""关闭事务消息生产者"""
if self.transaction_check_executor:
self.transaction_check_executor.shutdown(wait=True)
if self.producer:
self.producer.shutdown()
print("事务消息生产者关闭成功")
def set_transaction_listener(self, listener: TransactionListener):
"""设置事务监听器"""
self.transaction_listener = listener
def send_message_in_transaction(self, message: TransactionMessage,
arg: object = None) -> 'TransactionSendResult':
"""发送事务消息"""
try:
# 1. 发送半消息（Prepare消息）
prepare_result = self._send_prepare_message(message)
# 2. 执行本地事务
local_state = self.transaction_listener.execute_local_transaction(message, arg)
# 3. 根据本地事务结果提交或回滚
end_result = self._end_transaction(message, prepare_result, local_state)
# 4. 更新统计
self._update_transaction_stats(local_state)
return TransactionSendResult(
send_result=prepare_result,
local_transaction_state=local_state,
end_result=end_result
)
except Exception as e:
# 异常时回滚事务
if 'prepare_result' in locals():
self._end_transaction(message, prepare_result, LocalTransactionState.ROLLBACK_MESSAGE)
self._update_transaction_stats(LocalTransactionState.ROLLBACK_MESSAGE)
raise Exception(f"发送事务消息失败: {e}")
def _send_prepare_message(self, message: TransactionMessage) -> 'SendResult':
"""发送准备消息"""
# 转换为普通消息
normal_message = self._convert_to_message(message)
# 标记为准备消息
normal_message.put_property("PREPARE_MESSAGE", "true")
# 发送到Broker
result = self.producer.send(normal_message)
# 缓存事务状态
with self.lock:
self.transaction_states[message.transaction_id] = {
"message": message,
"prepare_result": result,
"state": LocalTransactionState.UNKNOWN,
"create_time": int(time.time() * 1000)
}
return result
def _end_transaction(self, message: TransactionMessage,
prepare_result: 'SendResult',
state: LocalTransactionState) -> bool:
"""结束事务"""
try:
# 发送事务结束请求到Broker
end_request = TransactionEndRequest(
transaction_id=message.transaction_id,
msg_id=prepare_result.msg_id,
transaction_state=state
)
# 这里应该发送到Broker
success = self._send_end_transaction_request(end_request)
# 更新缓存状态
with self.lock:
if message.transaction_id in self.transaction_states:
self.transaction_states[message.transaction_id]["state"] = state
return success
except Exception as e:
print(f"结束事务失败: {e}")
return False
def _send_end_transaction_request(self, request: 'TransactionEndRequest') -> bool:
"""发送事务结束请求"""
# 模拟发送到Broker
print(f"发送事务结束请求: {request.transaction_id}, 状态: {request.transaction_state}")
return True
def check_transaction_state(self, transaction_id: str) -> LocalTransactionState:
"""检查事务状态（Broker回查）"""
try:
with self.lock:
transaction_info = self.transaction_states.get(transaction_id)
if not transaction_info:
return LocalTransactionState.UNKNOWN
# 调用监听器检查本地事务状态
message = transaction_info["message"]
state = self.transaction_listener.check_local_transaction(message)
# 更新缓存状态
with self.lock:
transaction_info["state"] = state
return state
except Exception as e:
print(f"检查事务状态异常: {e}")
return LocalTransactionState.UNKNOWN
def _convert_to_message(self, transaction_message: TransactionMessage) -> 'Message':
"""转换为普通消息"""
from rocketmq_producer import Message
message = Message(
topic=transaction_message.topic,
tags=transaction_message.tags,
keys=transaction_message.keys,
body=transaction_message.body
)
# 复制属性
for key, value in transaction_message.properties.items():
message.put_property(key, value)
return message
def _update_transaction_stats(self, state: LocalTransactionState):
"""更新事务统计"""
with self.lock:
self.transaction_stats["total_transactions"] += 1
if state == LocalTransactionState.COMMIT_MESSAGE:
self.transaction_stats["committed_transactions"] += 1
elif state == LocalTransactionState.ROLLBACK_MESSAGE:
self.transaction_stats["rollback_transactions"] += 1
else:
self.transaction_stats["unknown_transactions"] += 1
def get_transaction_stats(self) -> Dict[str, int]:
"""获取事务统计"""
with self.lock:
return self.transaction_stats.copy()
def cleanup_expired_transactions(self, expire_time_ms: int = 300000):
"""清理过期事务"""
current_time = int(time.time() * 1000)
expired_transactions = []
with self.lock:
for transaction_id, info in self.transaction_states.items():
if current_time - info["create_time"] > expire_time_ms:
expired_transactions.append(transaction_id)
for transaction_id in expired_transactions:
del self.transaction_states[transaction_id]
print(f"清理过期事务: {len(expired_transactions)} 个")
# 事务发送结果
@dataclass
class TransactionSendResult:
send_result: 'SendResult'
local_transaction_state: LocalTransactionState
end_result: bool
# 事务结束请求
@dataclass
class TransactionEndRequest:
transaction_id: str
msg_id: str
transaction_state: LocalTransactionState
5.2.2 事务监听器实现
# 数据库事务监听器
class DatabaseTransactionListener(TransactionListener):
"""数据库事务监听器"""
def __init__(self, db_connection):
self.db_connection = db_connection
self.transaction_records = {} # 事务记录
self.lock = threading.Lock()
def execute_local_transaction(self, message: TransactionMessage,
arg: object) -> LocalTransactionState:
"""执行本地数据库事务"""
transaction_id = message.transaction_id
try:
# 开始数据库事务
self.db_connection.begin()
# 执行业务逻辑
success = self._execute_business_logic(message, arg)
if success:
# 提交数据库事务
self.db_connection.commit()
# 记录事务状态
with self.lock:
self.transaction_records[transaction_id] = {
"state": LocalTransactionState.COMMIT_MESSAGE,
"timestamp": int(time.time() * 1000),
"message": message
}
return LocalTransactionState.COMMIT_MESSAGE
else:
# 回滚数据库事务
self.db_connection.rollback()
with self.lock:
self.transaction_records[transaction_id] = {
"state": LocalTransactionState.ROLLBACK_MESSAGE,
"timestamp": int(time.time() * 1000),
"message": message
}
return LocalTransactionState.ROLLBACK_MESSAGE
except Exception as e:
print(f"执行本地事务异常: {e}")
# 回滚数据库事务
try:
self.db_connection.rollback()
except:
pass
with self.lock:
self.transaction_records[transaction_id] = {
"state": LocalTransactionState.ROLLBACK_MESSAGE,
"timestamp": int(time.time() * 1000),
"message": message,
"error": str(e)
}
return LocalTransactionState.ROLLBACK_MESSAGE
def check_local_transaction(self, message: TransactionMessage) -> LocalTransactionState:
"""检查本地事务状态"""
transaction_id = message.transaction_id
# 从缓存中查找
with self.lock:
record = self.transaction_records.get(transaction_id)
if record:
return record["state"]
# 从数据库中查找事务记录
try:
state = self._query_transaction_state_from_db(transaction_id)
return state
except Exception as e:
print(f"查询事务状态异常: {e}")
return LocalTransactionState.UNKNOWN
def _execute_business_logic(self, message: TransactionMessage, arg: object) -> bool:
"""执行业务逻辑"""
# 这里实现具体的业务逻辑
# 例如：更新订单状态、扣减库存等
try:
# 模拟业务操作
business_data = message.body.decode('utf-8')
print(f"执行业务逻辑: {business_data}")
# 执行SQL操作
cursor = self.db_connection.cursor()
cursor.execute(
"INSERT INTO transaction_log (transaction_id, message_data, create_time) VALUES (?, ?, ?)",
(message.transaction_id, business_data, int(time.time() * 1000))
)
return True
except Exception as e:
print(f"业务逻辑执行失败: {e}")
return False
def _query_transaction_state_from_db(self, transaction_id: str) -> LocalTransactionState:
"""从数据库查询事务状态"""
try:
cursor = self.db_connection.cursor()
cursor.execute(
"SELECT state FROM transaction_records WHERE transaction_id = ?",
(transaction_id,)
)
result = cursor.fetchone()
if result:
state_str = result[0]
return LocalTransactionState(state_str)
else:
return LocalTransactionState.UNKNOWN
except Exception as e:
print(f"查询数据库事务状态失败: {e}")
return LocalTransactionState.UNKNOWN
# 业务事务监听器
class BusinessTransactionListener(TransactionListener):
"""业务事务监听器"""
def __init__(self, business_executor: Callable[[TransactionMessage, object], bool]):
self.business_executor = business_executor
self.transaction_cache = {} # 事务缓存
self.lock = threading.Lock()
def execute_local_transaction(self, message: TransactionMessage,
arg: object) -> LocalTransactionState:
"""执行本地业务事务"""
transaction_id = message.transaction_id
try:
# 执行业务逻辑
success = self.business_executor(message, arg)
state = LocalTransactionState.COMMIT_MESSAGE if success else LocalTransactionState.ROLLBACK_MESSAGE
# 缓存事务状态
with self.lock:
self.transaction_cache[transaction_id] = {
"state": state,
"timestamp": int(time.time() * 1000),
"message": message
}
return state
except Exception as e:
print(f"执行业务事务异常: {e}")
with self.lock:
self.transaction_cache[transaction_id] = {
"state": LocalTransactionState.ROLLBACK_MESSAGE,
"timestamp": int(time.time() * 1000),
"message": message,
"error": str(e)
}
return LocalTransactionState.ROLLBACK_MESSAGE
def check_local_transaction(self, message: TransactionMessage) -> LocalTransactionState:
"""检查本地事务状态"""
transaction_id = message.transaction_id
with self.lock:
cache_record = self.transaction_cache.get(transaction_id)
if cache_record:
return cache_record["state"]
# 如果缓存中没有，返回未知状态
return LocalTransactionState.UNKNOWN
def get_transaction_records(self) -> Dict[str, Dict]:
"""获取事务记录"""
with self.lock:
return self.transaction_cache.copy()
def cleanup_expired_records(self, expire_time_ms: int = 600000):
"""清理过期记录"""
current_time = int(time.time() * 1000)
expired_keys = []
with self.lock:
for transaction_id, record in self.transaction_cache.items():
if current_time - record["timestamp"] > expire_time_ms:
expired_keys.append(transaction_id)
for key in expired_keys:
del self.transaction_cache[key]
print(f"清理过期事务记录: {len(expired_keys)} 个")
5.3 延时消息
5.3.1 延时消息实现
# 延时级别枚举
class DelayLevel(Enum):
LEVEL_1 = 1 # 1s
LEVEL_2 = 2 # 5s
LEVEL_3 = 3 # 10s
LEVEL_4 = 4 # 30s
LEVEL_5 = 5 # 1m
LEVEL_6 = 6 # 2m
LEVEL_7 = 7 # 3m
LEVEL_8 = 8 # 4m
LEVEL_9 = 9 # 5m
LEVEL_10 = 10 # 6m
LEVEL_11 = 11 # 7m
LEVEL_12 = 12 # 8m
LEVEL_13 = 13 # 9m
LEVEL_14 = 14 # 10m
LEVEL_15 = 15 # 20m
LEVEL_16 = 16 # 30m
LEVEL_17 = 17 # 1h
LEVEL_18 = 18 # 2h
# 延时消息
@dataclass
class DelayMessage:
topic: str
tags: str
keys: str
body: bytes
delay_level: DelayLevel
properties: Dict[str, str] = None
def __post_init__(self):
if self.properties is None:
self.properties = {}
# 设置延时级别
self.properties["DELAY"] = str(self.delay_level.value)
def get_delay_time_ms(self) -> int:
"""获取延时时间（毫秒）"""
delay_times = {
DelayLevel.LEVEL_1: 1000,
DelayLevel.LEVEL_2: 5000,
DelayLevel.LEVEL_3: 10000,
DelayLevel.LEVEL_4: 30000,
DelayLevel.LEVEL_5: 60000,
DelayLevel.LEVEL_6: 120000,
DelayLevel.LEVEL_7: 180000,
DelayLevel.LEVEL_8: 240000,
DelayLevel.LEVEL_9: 300000,
DelayLevel.LEVEL_10: 360000,
DelayLevel.LEVEL_11: 420000,
DelayLevel.LEVEL_12: 480000,
DelayLevel.LEVEL_13: 540000,
DelayLevel.LEVEL_14: 600000,
DelayLevel.LEVEL_15: 1200000,
DelayLevel.LEVEL_16: 1800000,
DelayLevel.LEVEL_17: 3600000,
DelayLevel.LEVEL_18: 7200000,
}
return delay_times.get(self.delay_level, 1000)
# 延时消息生产者
class DelayMessageProducer:
"""延时消息生产者"""
def __init__(self, producer_config: 'ProducerConfig'):
self.config = producer_config
self.producer = None
self.delay_stats = {
"total_sent": 0,
"success_sent": 0,
"failed_sent": 0,
"delay_levels": {level.value: 0 for level in DelayLevel}
}
def start(self):
"""启动延时消息生产者"""
from rocketmq_producer import DefaultMQProducer
self.producer = DefaultMQProducer(self.config)
self.producer.start()
print("延时消息生产者启动成功")
def shutdown(self):
"""关闭延时消息生产者"""
if self.producer:
self.producer.shutdown()
print("延时消息生产者关闭成功")
def send_delay_message(self, message: DelayMessage) -> 'SendResult':
"""发送延时消息"""
try:
# 转换为普通消息
normal_message = self._convert_to_message(message)
# 发送消息
result = self.producer.send(normal_message)
# 更新统计
self.delay_stats["total_sent"] += 1
self.delay_stats["success_sent"] += 1
self.delay_stats["delay_levels"][message.delay_level.value] += 1
print(f"发送延时消息成功: {result.msg_id}, 延时级别: {message.delay_level}")
return result
except Exception as e:
self.delay_stats["total_sent"] += 1
self.delay_stats["failed_sent"] += 1
raise Exception(f"发送延时消息失败: {e}")
def send_delay_message_async(self, message: DelayMessage,
callback: 'SendCallback'):
"""异步发送延时消息"""
try:
normal_message = self._convert_to_message(message)
# 包装回调
wrapped_callback = DelayMessageSendCallback(
callback, self.delay_stats, message.delay_level
)
self.producer.send_async(normal_message, wrapped_callback)
except Exception as e:
self.delay_stats["total_sent"] += 1
self.delay_stats["failed_sent"] += 1
callback.on_exception(e)
def send_delay_message_at_time(self, message: DelayMessage,
deliver_time: int) -> 'SendResult':
"""在指定时间发送延时消息"""
current_time = int(time.time() * 1000)
delay_time = deliver_time - current_time
if delay_time <= 0:
# 立即发送
normal_message = self._convert_to_message_without_delay(message)
return self.producer.send(normal_message)
# 计算最接近的延时级别
delay_level = self._calculate_delay_level(delay_time)
message.delay_level = delay_level
return self.send_delay_message(message)
def _convert_to_message(self, delay_message: DelayMessage) -> 'Message':
"""转换为普通消息"""
from rocketmq_producer import Message
message = Message(
topic=delay_message.topic,
tags=delay_message.tags,
keys=delay_message.keys,
body=delay_message.body
)
# 设置延时级别
message.put_property("DELAY", str(delay_message.delay_level.value))
# 复制其他属性
for key, value in delay_message.properties.items():
if key != "DELAY":
message.put_property(key, value)
return message
def _convert_to_message_without_delay(self, delay_message: DelayMessage) -> 'Message':
"""转换为普通消息（不设置延时）"""
from rocketmq_producer import Message
message = Message(
topic=delay_message.topic,
tags=delay_message.tags,
keys=delay_message.keys,
body=delay_message.body
)
# 复制属性（排除延时属性）
for key, value in delay_message.properties.items():
if key != "DELAY":
message.put_property(key, value)
return message
def _calculate_delay_level(self, delay_time_ms: int) -> DelayLevel:
"""计算最接近的延时级别"""
delay_times = [
(DelayLevel.LEVEL_1, 1000),
(DelayLevel.LEVEL_2, 5000),
(DelayLevel.LEVEL_3, 10000),
(DelayLevel.LEVEL_4, 30000),
(DelayLevel.LEVEL_5, 60000),
(DelayLevel.LEVEL_6, 120000),
(DelayLevel.LEVEL_7, 180000),
(DelayLevel.LEVEL_8, 240000),
(DelayLevel.LEVEL_9, 300000),
(DelayLevel.LEVEL_10, 360000),
(DelayLevel.LEVEL_11, 420000),
(DelayLevel.LEVEL_12, 480000),
(DelayLevel.LEVEL_13, 540000),
(DelayLevel.LEVEL_14, 600000),
(DelayLevel.LEVEL_15, 1200000),
(DelayLevel.LEVEL_16, 1800000),
(DelayLevel.LEVEL_17, 3600000),
(DelayLevel.LEVEL_18, 7200000),
]
# 找到最接近的延时级别
best_level = DelayLevel.LEVEL_1
min_diff = abs(delay_time_ms - 1000)
for level, time_ms in delay_times:
diff = abs(delay_time_ms - time_ms)
if diff < min_diff:
min_diff = diff
best_level = level
return best_level
def get_delay_stats(self) -> Dict:
"""获取延时消息统计"""
return self.delay_stats.copy()
# 延时消息发送回调
class DelayMessageSendCallback:
"""延时消息发送回调"""
def __init__(self, callback: 'SendCallback', stats: Dict, delay_level: DelayLevel):
self.callback = callback
self.stats = stats
self.delay_level = delay_level
def on_success(self, result: 'SendResult'):
"""发送成功回调"""
self.stats["total_sent"] += 1
self.stats["success_sent"] += 1
self.stats["delay_levels"][self.delay_level.value] += 1
self.callback.on_success(result)
def on_exception(self, exception: Exception):
"""发送异常回调"""
self.stats["total_sent"] += 1
self.stats["failed_sent"] += 1
self.callback.on_exception(exception)
5.3.2 延时消息使用示例
# 延时消息使用示例
class DelayMessageExample:
"""延时消息使用示例"""
def __init__(self):
# 配置生产者
self.producer_config = ProducerConfig(
producer_group="delay_producer_group",
name_server_addr="localhost:9876"
)
# 创建延时消息生产者
self.producer = DelayMessageProducer(self.producer_config)
def start_example(self):
"""启动示例"""
try:
# 启动生产者
self.producer.start()
# 发送不同延时级别的消息
self._send_various_delay_messages()
# 发送定时消息
self._send_scheduled_messages()
# 打印统计信息
self._print_statistics()
finally:
self.producer.shutdown()
def _send_various_delay_messages(self):
"""发送不同延时级别的消息"""
delay_levels = [
DelayLevel.LEVEL_1, # 1秒
DelayLevel.LEVEL_3, # 10秒
DelayLevel.LEVEL_5, # 1分钟
DelayLevel.LEVEL_15, # 20分钟
DelayLevel.LEVEL_17 # 1小时
]
for i, delay_level in enumerate(delay_levels):
message = DelayMessage(
topic="DelayTopic",
tags="DelayTag",
keys=f"delay_key_{i}",
body=f"延时消息内容 {i}, 延时级别: {delay_level.name}".encode('utf-8'),
delay_level=delay_level
)
try:
result = self.producer.send_delay_message(message)
print(f"发送延时消息成功: {result.msg_id}")
except Exception as e:
print(f"发送延时消息失败: {e}")
def _send_scheduled_messages(self):
"""发送定时消息"""
# 5分钟后发送
deliver_time = int(time.time() * 1000) + 5 * 60 * 1000
message = DelayMessage(
topic="DelayTopic",
tags="ScheduledTag",
keys="scheduled_key",
body="定时消息内容".encode('utf-8'),
delay_level=DelayLevel.LEVEL_1 # 会被重新计算
)
try:
result = self.producer.send_delay_message_at_time(message, deliver_time)
print(f"发送定时消息成功: {result.msg_id}")
except Exception as e:
print(f"发送定时消息失败: {e}")
def _print_statistics(self):
"""打印统计信息"""
stats = self.producer.get_delay_stats()
print("\n延时消息统计:")
print(f"总发送数: {stats['total_sent']}")
print(f"成功发送数: {stats['success_sent']}")
print(f"失败发送数: {stats['failed_sent']}")
print("\n各延时级别统计:")
for level, count in stats['delay_levels'].items():
if count > 0:
print(f" 级别 {level}: {count} 条")
# 延时消息消费者示例
class DelayMessageConsumerExample:
"""延时消息消费者示例"""
def __init__(self):
# 配置消费者
self.consumer_config = ConsumerConfig(
consumer_group="delay_consumer_group",
name_server_addr="localhost:9876"
)
# 创建消费者
from rocketmq_consumer import DefaultMQPushConsumer
self.consumer = DefaultMQPushConsumer(self.consumer_config)
def start_consume(self):
"""开始消费延时消息"""
try:
# 注册消息监听器
listener = DelayMessageListener()
self.consumer.register_message_listener(listener)
# 订阅主题
self.consumer.subscribe("DelayTopic", "*")
# 启动消费者
self.consumer.start()
print("延时消息消费者启动成功，等待消息...")
# 等待消息
while True:
time.sleep(10)
stats = listener.get_consume_stats()
print(f"消费统计: {stats}")
except KeyboardInterrupt:
print("收到中断信号，正在关闭消费者...")
finally:
self.consumer.shutdown()
# 延时消息监听器
class DelayMessageListener(MessageListenerConcurrently):
"""延时消息监听器"""
def __init__(self):
self.consume_stats = {
"total_consumed": 0,
"success_consumed": 0,
"failed_consumed": 0,
"delay_levels": {}
}
def consume_message(self, messages: List['MessageExt'],
context: 'ConsumeContext') -> 'ConsumeResult':
"""消费延时消息"""
for msg in messages:
try:
# 获取延时级别
delay_level = msg.get_property("DELAY")
# 计算实际延时时间
actual_delay = self._calculate_actual_delay(msg)
print(f"收到延时消息:")
print(f" 消息ID: {msg.msg_id}")
print(f" 延时级别: {delay_level}")
print(f" 实际延时: {actual_delay}ms")
print(f" 消息内容: {msg.body.decode('utf-8')}")
# 更新统计
self.consume_stats["total_consumed"] += 1
self.consume_stats["success_consumed"] += 1
if delay_level:
level_stats = self.consume_stats["delay_levels"]
level_stats[delay_level] = level_stats.get(delay_level, 0) + 1
except Exception as e:
print(f"处理延时消息异常: {e}")
self.consume_stats["total_consumed"] += 1
self.consume_stats["failed_consumed"] += 1
return ConsumeResult.RECONSUME_LATER
return ConsumeResult.SUCCESS
def _calculate_actual_delay(self, msg: 'MessageExt') -> int:
"""计算实际延时时间"""
try:
# 消息存储时间 - 消息产生时间
return msg.store_timestamp - msg.born_timestamp
except:
return 0
def get_consume_stats(self) -> Dict:
"""获取消费统计"""
return self.consume_stats.copy()
# 使用示例
if __name__ == "__main__":
# 发送延时消息示例
producer_example = DelayMessageExample()
producer_example.start_example()
# 消费延时消息示例
# consumer_example = DelayMessageConsumerExample()
# consumer_example.start_consume()
5.4 批量消息
5.4.1 批量消息实现
# 批量消息
@dataclass
class BatchMessage:
topic: str
messages: List['Message']
def __post_init__(self):
if not self.messages:
raise ValueError("批量消息不能为空")
# 验证所有消息的主题一致
for msg in self.messages:
if msg.topic != self.topic:
raise ValueError(f"消息主题不一致: {msg.topic} != {self.topic}")
def get_total_size(self) -> int:
"""获取批量消息总大小"""
total_size = 0
for msg in self.messages:
total_size += len(msg.body)
total_size += len(msg.topic.encode('utf-8'))
total_size += len(msg.tags.encode('utf-8')) if msg.tags else 0
total_size += len(msg.keys.encode('utf-8')) if msg.keys else 0
# 属性大小
for key, value in msg.properties.items():
total_size += len(key.encode('utf-8'))
total_size += len(value.encode('utf-8'))
return total_size
def get_message_count(self) -> int:
"""获取消息数量"""
return len(self.messages)
# 批量消息生产者
class BatchMessageProducer:
"""批量消息生产者"""
def __init__(self, producer_config: 'ProducerConfig'):
self.config = producer_config
self.producer = None
self.max_batch_size = 1024 * 1024 # 1MB
self.max_batch_count = 1000
self.batch_stats = {
"total_batches": 0,
"total_messages": 0,
"success_batches": 0,
"failed_batches": 0,
"avg_batch_size": 0
}
self.batch_sizes = []
def start(self):
"""启动批量消息生产者"""
from rocketmq_producer import DefaultMQProducer
self.producer = DefaultMQProducer(self.config)
self.producer.start()
print("批量消息生产者启动成功")
def shutdown(self):
"""关闭批量消息生产者"""
if self.producer:
self.producer.shutdown()
print("批量消息生产者关闭成功")
def send_batch(self, batch_message: BatchMessage) -> 'SendResult':
"""发送批量消息"""
try:
# 验证批量消息
self._validate_batch_message(batch_message)
# 如果批量过大，分割发送
if self._need_split_batch(batch_message):
return self._send_split_batch(batch_message)
# 发送单个批量
result = self._send_single_batch(batch_message)
# 更新统计
self._update_batch_stats(batch_message, True)
return result
except Exception as e:
self._update_batch_stats(batch_message, False)
raise Exception(f"发送批量消息失败: {e}")
def send_batch_async(self, batch_message: BatchMessage,
callback: 'BatchSendCallback'):
"""异步发送批量消息"""
try:
self._validate_batch_message(batch_message)
if self._need_split_batch(batch_message):
self._send_split_batch_async(batch_message, callback)
else:
self._send_single_batch_async(batch_message, callback)
except Exception as e:
self._update_batch_stats(batch_message, False)
callback.on_exception(e)
def _validate_batch_message(self, batch_message: BatchMessage):
"""验证批量消息"""
if not batch_message.messages:
raise ValueError("批量消息不能为空")
if len(batch_message.messages) > self.max_batch_count:
raise ValueError(f"批量消息数量超限: {len(batch_message.messages)} > {self.max_batch_count}")
# 检查消息是否包含延时或事务属性
for msg in batch_message.messages:
if msg.get_property("DELAY") or msg.get_property("TRAN_MSG"):
raise ValueError("批量消息不支持延时或事务消息")
def _need_split_batch(self, batch_message: BatchMessage) -> bool:
"""判断是否需要分割批量"""
return batch_message.get_total_size() > self.max_batch_size
def _send_single_batch(self, batch_message: BatchMessage) -> 'SendResult':
"""发送单个批量消息"""
# 将批量消息编码为单个消息
encoded_message = self._encode_batch_message(batch_message)
# 发送到Broker
result = self.producer.send(encoded_message)
print(f"发送批量消息成功: {result.msg_id}, 消息数量: {batch_message.get_message_count()}")
return result
def _send_single_batch_async(self, batch_message: BatchMessage,
callback: 'BatchSendCallback'):
"""异步发送单个批量消息"""
encoded_message = self._encode_batch_message(batch_message)
# 包装回调
wrapped_callback = BatchSendCallbackWrapper(
callback, self.batch_stats, batch_message
)
self.producer.send_async(encoded_message, wrapped_callback)
def _send_split_batch(self, batch_message: BatchMessage) -> 'SendResult':
"""分割发送批量消息"""
split_batches = self._split_batch_message(batch_message)
results = []
for split_batch in split_batches:
result = self._send_single_batch(split_batch)
results.append(result)
# 返回第一个结果（可以考虑返回聚合结果）
return results[0] if results else None
def _send_split_batch_async(self, batch_message: BatchMessage,
callback: 'BatchSendCallback'):
"""异步分割发送批量消息"""
split_batches = self._split_batch_message(batch_message)
# 创建聚合回调
aggregated_callback = AggregatedBatchSendCallback(
callback, len(split_batches)
)
for split_batch in split_batches:
self._send_single_batch_async(split_batch, aggregated_callback)
def _split_batch_message(self, batch_message: BatchMessage) -> List[BatchMessage]:
"""分割批量消息"""
split_batches = []
current_batch = []
current_size = 0
for msg in batch_message.messages:
msg_size = self._calculate_message_size(msg)
if (current_size + msg_size > self.max_batch_size or
len(current_batch) >= self.max_batch_count) and current_batch:
# 创建新的批量
split_batches.append(BatchMessage(
topic=batch_message.topic,
messages=current_batch.copy()
))
current_batch = []
current_size = 0
current_batch.append(msg)
current_size += msg_size
# 添加最后一个批量
if current_batch:
split_batches.append(BatchMessage(
topic=batch_message.topic,
messages=current_batch
))
return split_batches
def _encode_batch_message(self, batch_message: BatchMessage) -> 'Message':
"""编码批量消息"""
from rocketmq_producer import Message
import json
# 将批量消息序列化
batch_data = {
"topic": batch_message.topic,
"messages": []
}
for msg in batch_message.messages:
msg_data = {
"tags": msg.tags,
"keys": msg.keys,
"body": msg.body.decode('utf-8') if isinstance(msg.body, bytes) else msg.body,
"properties": msg.properties
}
batch_data["messages"].append(msg_data)
# 创建编码后的消息
encoded_message = Message(
topic=batch_message.topic,
tags="BATCH",
keys=f"batch_{int(time.time() * 1000)}",
body=json.dumps(batch_data).encode('utf-8')
)
# 设置批量标识
encoded_message.put_property("BATCH_MESSAGE", "true")
encoded_message.put_property("BATCH_SIZE", str(batch_message.get_message_count()))
return encoded_message
def _calculate_message_size(self, message: 'Message') -> int:
"""计算单个消息大小"""
size = len(message.body)
size += len(message.topic.encode('utf-8'))
size += len(message.tags.encode('utf-8')) if message.tags else 0
size += len(message.keys.encode('utf-8')) if message.keys else 0
for key, value in message.properties.items():
size += len(key.encode('utf-8'))
size += len(value.encode('utf-8'))
return size
def _update_batch_stats(self, batch_message: BatchMessage, success: bool):
"""更新批量统计"""
self.batch_stats["total_batches"] += 1
self.batch_stats["total_messages"] += batch_message.get_message_count()
if success:
self.batch_stats["success_batches"] += 1
else:
self.batch_stats["failed_batches"] += 1
# 记录批量大小
batch_size = batch_message.get_total_size()
self.batch_sizes.append(batch_size)
# 计算平均批量大小
if self.batch_sizes:
self.batch_stats["avg_batch_size"] = sum(self.batch_sizes) // len(self.batch_sizes)
def get_batch_stats(self) -> Dict:
"""获取批量统计"""
return self.batch_stats.copy()
def set_max_batch_size(self, max_size: int):
"""设置最大批量大小"""
self.max_batch_size = max_size
def set_max_batch_count(self, max_count: int):
"""设置最大批量数量"""
self.max_batch_count = max_count
# 批量发送回调接口
class BatchSendCallback(ABC):
"""批量发送回调接口"""
@abstractmethod
def on_success(self, result: 'SendResult', batch_message: BatchMessage):
"""批量发送成功回调"""
pass
@abstractmethod
def on_exception(self, exception: Exception):
"""批量发送异常回调"""
pass
# 批量发送回调包装器
class BatchSendCallbackWrapper:
"""批量发送回调包装器"""
def __init__(self, callback: BatchSendCallback, stats: Dict, batch_message: BatchMessage):
self.callback = callback
self.stats = stats
self.batch_message = batch_message
def on_success(self, result: 'SendResult'):
"""发送成功回调"""
# 更新统计
self.stats["total_batches"] += 1
self.stats["success_batches"] += 1
self.stats["total_messages"] += self.batch_message.get_message_count()
self.callback.on_success(result, self.batch_message)
def on_exception(self, exception: Exception):
"""发送异常回调"""
# 更新统计
self.stats["total_batches"] += 1
self.stats["failed_batches"] += 1
self.stats["total_messages"] += self.batch_message.get_message_count()
self.callback.on_exception(exception)
# 聚合批量发送回调
class AggregatedBatchSendCallback(BatchSendCallback):
"""聚合批量发送回调"""
def __init__(self, original_callback: BatchSendCallback, total_batches: int):
self.original_callback = original_callback
self.total_batches = total_batches
self.completed_batches = 0
self.success_results = []
self.exceptions = []
self.lock = threading.Lock()
def on_success(self, result: 'SendResult', batch_message: BatchMessage):
"""批量发送成功回调"""
with self.lock:
self.completed_batches += 1
self.success_results.append((result, batch_message))
if self.completed_batches == self.total_batches:
# 所有批量都完成了
if self.exceptions:
# 有异常发生
self.original_callback.on_exception(self.exceptions[0])
else:
# 全部成功
first_result, first_batch = self.success_results[0]
self.original_callback.on_success(first_result, first_batch)
def on_exception(self, exception: Exception):
"""批量发送异常回调"""
with self.lock:
self.completed_batches += 1
self.exceptions.append(exception)
if self.completed_batches == self.total_batches:
# 所有批量都完成了
self.original_callback.on_exception(self.exceptions[0])
### 5.4.2 批量消息消费
```python
# 批量消息消费者
class BatchMessageConsumer:
"""批量消息消费者"""
def __init__(self, consumer_config: 'ConsumerConfig'):
self.config = consumer_config
self.consumer = None
self.batch_listener: Optional['BatchMessageListener'] = None
self.consume_stats = {
"total_batches": 0,
"total_messages": 0,
"success_batches": 0,
"failed_batches": 0
}
def start(self):
"""启动批量消息消费者"""
if not self.batch_listener:
raise Exception("未设置批量消息监听器")
from rocketmq_consumer import DefaultMQPushConsumer
self.consumer = DefaultMQPushConsumer(self.config)
# 注册批量消息监听器
wrapped_listener = BatchMessageListenerWrapper(
self.batch_listener, self.consume_stats
)
self.consumer.register_message_listener(wrapped_listener)
self.consumer.start()
print("批量消息消费者启动成功")
def shutdown(self):
"""关闭批量消息消费者"""
if self.consumer:
self.consumer.shutdown()
print("批量消息消费者关闭成功")
def subscribe(self, topic: str, sub_expression: str = "*"):
"""订阅主题"""
if self.consumer:
self.consumer.subscribe(topic, sub_expression)
def register_batch_listener(self, listener: 'BatchMessageListener'):
"""注册批量消息监听器"""
self.batch_listener = listener
def get_consume_stats(self) -> Dict[str, int]:
"""获取消费统计"""
return self.consume_stats.copy()
# 批量消息监听器接口
class BatchMessageListener(ABC):
"""批量消息监听器接口"""
@abstractmethod
def consume_batch_message(self, batch_messages: List['MessageExt']) -> 'ConsumeResult':
"""消费批量消息"""
pass
# 批量消息监听器包装器
class BatchMessageListenerWrapper:
"""批量消息监听器包装器"""
def __init__(self, batch_listener: BatchMessageListener, stats: Dict[str, int]):
self.batch_listener = batch_listener
self.stats = stats
def consume_message(self, messages: List['MessageExt'],
context: 'ConsumeContext') -> 'ConsumeResult':
"""消费消息（适配器方法）"""
try:
# 检查是否为批量消息
batch_messages = []
individual_messages = []
for msg in messages:
if msg.get_property("BATCH_MESSAGE") == "true":
# 解码批量消息
decoded_messages = self._decode_batch_message(msg)
batch_messages.extend(decoded_messages)
else:
individual_messages.append(msg)
# 处理批量消息
if batch_messages:
result = self.batch_listener.consume_batch_message(batch_messages)
self._update_batch_stats(len(batch_messages), result == ConsumeResult.SUCCESS)
return result
# 处理单个消息
if individual_messages:
result = self.batch_listener.consume_batch_message(individual_messages)
self._update_individual_stats(len(individual_messages), result == ConsumeResult.SUCCESS)
return result
return ConsumeResult.SUCCESS
except Exception as e:
print(f"批量消费异常: {e}")
self._update_batch_stats(len(messages), False)
return ConsumeResult.RECONSUME_LATER
def _decode_batch_message(self, batch_msg: 'MessageExt') -> List['MessageExt']:
"""解码批量消息"""
import json
try:
# 解析批量消息数据
batch_data = json.loads(batch_msg.body.decode('utf-8'))
decoded_messages = []
for msg_data in batch_data["messages"]:
# 创建消息扩展对象
decoded_msg = MessageExt(
topic=batch_data["topic"],
tags=msg_data["tags"],
keys=msg_data["keys"],
body=msg_data["body"].encode('utf-8'),
msg_id=f"{batch_msg.msg_id}_{len(decoded_messages)}",
queue_id=batch_msg.queue_id,
queue_offset=batch_msg.queue_offset,
born_timestamp=batch_msg.born_timestamp,
store_timestamp=batch_msg.store_timestamp
)
# 设置属性
for key, value in msg_data["properties"].items():
decoded_msg.put_property(key, value)
decoded_messages.append(decoded_msg)
return decoded_messages
except Exception as e:
print(f"解码批量消息失败: {e}")
return []
def _update_batch_stats(self, message_count: int, success: bool):
"""更新批量统计"""
self.stats["total_batches"] += 1
self.stats["total_messages"] += message_count
if success:
self.stats["success_batches"] += 1
else:
self.stats["failed_batches"] += 1
def _update_individual_stats(self, message_count: int, success: bool):
"""更新单个消息统计"""
self.stats["total_messages"] += message_count
if success:
self.stats["success_batches"] += 1
else:
self.stats["failed_batches"] += 1
# 业务批量消息监听器
class BusinessBatchMessageListener(BatchMessageListener):
"""业务批量消息监听器"""
def __init__(self, batch_processor: Callable[[List['MessageExt']], bool]):
self.batch_processor = batch_processor
self.processed_count = 0
self.failed_count = 0
def consume_batch_message(self, batch_messages: List['MessageExt']) -> 'ConsumeResult':
"""消费批量消息"""
try:
print(f"收到批量消息: {len(batch_messages)} 条")
# 处理批量消息
success = self.batch_processor(batch_messages)
if success:
self.processed_count += len(batch_messages)
print(f"批量消息处理成功: {len(batch_messages)} 条")
return ConsumeResult.SUCCESS
else:
self.failed_count += len(batch_messages)
print(f"批量消息处理失败: {len(batch_messages)} 条")
return ConsumeResult.RECONSUME_LATER
except Exception as e:
print(f"批量消息处理异常: {e}")
self.failed_count += len(batch_messages)
return ConsumeResult.RECONSUME_LATER
def get_process_stats(self) -> Dict[str, int]:
"""获取处理统计"""
return {
"processed_count": self.processed_count,
"failed_count": self.failed_count,
"total_count": self.processed_count + self.failed_count
}
5.4.3 批量消息使用示例
# 批量消息使用示例
class BatchMessageExample:
"""批量消息使用示例"""
def __init__(self):
# 配置生产者
self.producer_config = ProducerConfig(
producer_group="batch_producer_group",
name_server_addr="localhost:9876"
)
# 配置消费者
self.consumer_config = ConsumerConfig(
consumer_group="batch_consumer_group",
name_server_addr="localhost:9876"
)
# 创建生产者和消费者
self.producer = BatchMessageProducer(self.producer_config)
self.consumer = BatchMessageConsumer(self.consumer_config)
def start_producer_example(self):
"""启动生产者示例"""
try:
self.producer.start()
# 发送批量消息
self._send_batch_messages()
# 发送大批量消息（会自动分割）
self._send_large_batch_messages()
# 异步发送批量消息
self._send_batch_messages_async()
# 打印统计信息
self._print_producer_stats()
finally:
self.producer.shutdown()
def start_consumer_example(self):
"""启动消费者示例"""
try:
# 注册批量消息监听器
listener = BusinessBatchMessageListener(self._process_batch_messages)
self.consumer.register_batch_listener(listener)
# 订阅主题
self.consumer.subscribe("BatchTopic", "*")
# 启动消费者
self.consumer.start()
print("批量消息消费者启动成功，等待消息...")
# 等待消息
while True:
time.sleep(10)
stats = self.consumer.get_consume_stats()
process_stats = listener.get_process_stats()
print(f"消费统计: {stats}")
print(f"处理统计: {process_stats}")
except KeyboardInterrupt:
print("收到中断信号，正在关闭消费者...")
finally:
self.consumer.shutdown()
def _send_batch_messages(self):
"""发送批量消息"""
from rocketmq_producer import Message
# 创建批量消息
messages = []
for i in range(10):
message = Message(
topic="BatchTopic",
tags="BatchTag",
keys=f"batch_key_{i}",
body=f"批量消息内容 {i}".encode('utf-8')
)
messages.append(message)
batch_message = BatchMessage(
topic="BatchTopic",
messages=messages
)
try:
result = self.producer.send_batch(batch_message)
print(f"发送批量消息成功: {result.msg_id}")
except Exception as e:
print(f"发送批量消息失败: {e}")
def _send_large_batch_messages(self):
"""发送大批量消息（会自动分割）"""
from rocketmq_producer import Message
# 创建大批量消息
messages = []
for i in range(1500): # 超过最大批量数量
message = Message(
topic="BatchTopic",
tags="LargeBatchTag",
keys=f"large_batch_key_{i}",
body=f"大批量消息内容 {i} - {'X' * 1000}".encode('utf-8') # 增加消息大小
)
messages.append(message)
batch_message = BatchMessage(
topic="BatchTopic",
messages=messages
)
try:
result = self.producer.send_batch(batch_message)
print(f"发送大批量消息成功: {result.msg_id}")
except Exception as e:
print(f"发送大批量消息失败: {e}")
def _send_batch_messages_async(self):
"""异步发送批量消息"""
from rocketmq_producer import Message
# 创建批量消息
messages = []
for i in range(5):
message = Message(
topic="BatchTopic",
tags="AsyncBatchTag",
keys=f"async_batch_key_{i}",
body=f"异步批量消息内容 {i}".encode('utf-8')
)
messages.append(message)
batch_message = BatchMessage(
topic="BatchTopic",
messages=messages
)
# 创建回调
callback = AsyncBatchSendCallback()
try:
self.producer.send_batch_async(batch_message, callback)
print("异步发送批量消息请求已提交")
# 等待回调完成
time.sleep(2)
except Exception as e:
print(f"异步发送批量消息失败: {e}")
def _process_batch_messages(self, messages: List['MessageExt']) -> bool:
"""处理批量消息"""
try:
print(f"\n处理批量消息: {len(messages)} 条")
for i, msg in enumerate(messages):
print(f" 消息 {i+1}:")
print(f" ID: {msg.msg_id}")
print(f" Tags: {msg.tags}")
print(f" Keys: {msg.keys}")
print(f" 内容: {msg.body.decode('utf-8')[:50]}...")
# 模拟批量处理
time.sleep(0.1 * len(messages))
print(f"批量消息处理完成: {len(messages)} 条")
return True
except Exception as e:
print(f"批量消息处理异常: {e}")
return False
def _print_producer_stats(self):
"""打印生产者统计信息"""
stats = self.producer.get_batch_stats()
print("\n批量消息生产者统计:")
print(f"总批量数: {stats['total_batches']}")
print(f"总消息数: {stats['total_messages']}")
print(f"成功批量数: {stats['success_batches']}")
print(f"失败批量数: {stats['failed_batches']}")
print(f"平均批量大小: {stats['avg_batch_size']} 字节")
# 异步批量发送回调
class AsyncBatchSendCallback(BatchSendCallback):
"""异步批量发送回调"""
def on_success(self, result: 'SendResult', batch_message: BatchMessage):
"""批量发送成功回调"""
print(f"异步批量发送成功: {result.msg_id}, 消息数量: {batch_message.get_message_count()}")
def on_exception(self, exception: Exception):
"""批量发送异常回调"""
print(f"异步批量发送失败: {exception}")
# 使用示例
if __name__ == "__main__":
example = BatchMessageExample()
# 发送批量消息示例
example.start_producer_example()
# 消费批量消息示例
# example.start_consumer_example()
5.5 消息过滤
5.5.1 标签过滤
# 标签过滤枚举
class FilterType(Enum):
"""过滤类型枚举"""
TAG = "TAG" # 标签过滤
SQL92 = "SQL92" # SQL92过滤
# 消息过滤器接口
class MessageFilter(ABC):
"""消息过滤器接口"""
@abstractmethod
def match(self, message: 'MessageExt') -> bool:
"""检查消息是否匹配过滤条件"""
pass
@abstractmethod
def get_filter_expression(self) -> str:
"""获取过滤表达式"""
pass
@abstractmethod
def get_filter_type(self) -> FilterType:
"""获取过滤类型"""
pass
# 标签过滤器
class TagFilter(MessageFilter):
"""标签过滤器"""
def __init__(self, tag_expression: str):
self.tag_expression = tag_expression
self.tags = self._parse_tag_expression(tag_expression)
def match(self, message: 'MessageExt') -> bool:
"""检查消息标签是否匹配"""
if not message.tags:
return "*" in self.tags
# 支持通配符
if "*" in self.tags:
return True
# 精确匹配
return message.tags in self.tags
def get_filter_expression(self) -> str:
"""获取过滤表达式"""
return self.tag_expression
def get_filter_type(self) -> FilterType:
"""获取过滤类型"""
return FilterType.TAG
def _parse_tag_expression(self, expression: str) -> Set[str]:
"""解析标签表达式"""
if not expression or expression == "*":
return {"*"}
# 支持多个标签，用 || 分隔
tags = set()
for tag in expression.split("||"):
tag = tag.strip()
if tag:
tags.add(tag)
return tags if tags else {"*"}
# SQL92过滤器
class SQL92Filter(MessageFilter):
"""SQL92过滤器"""
def __init__(self, sql_expression: str):
self.sql_expression = sql_expression
self.compiled_expression = self._compile_expression(sql_expression)
def match(self, message: 'MessageExt') -> bool:
"""检查消息是否匹配SQL表达式"""
try:
# 构建消息上下文
context = self._build_message_context(message)
# 执行SQL表达式
return self._evaluate_expression(self.compiled_expression, context)
except Exception as e:
print(f"SQL过滤器执行异常: {e}")
return False
def get_filter_expression(self) -> str:
"""获取过滤表达式"""
return self.sql_expression
def get_filter_type(self) -> FilterType:
"""获取过滤类型"""
return FilterType.SQL92
def _compile_expression(self, expression: str) -> Dict[str, Any]:
"""编译SQL表达式"""
# 简化的SQL解析器
import re
compiled = {
"original": expression,
"conditions": []
}
# 解析WHERE条件
where_match = re.search(r'WHERE\s+(.+)', expression, re.IGNORECASE)
if where_match:
where_clause = where_match.group(1)
# 解析条件（简化版本）
conditions = self._parse_where_clause(where_clause)
compiled["conditions"] = conditions
return compiled
def _parse_where_clause(self, where_clause: str) -> List[Dict[str, Any]]:
"""解析WHERE子句"""
import re
conditions = []
# 支持的操作符
operators = ['=', '!=', '>', '<', '>=', '<=', 'LIKE', 'IN']
# 简单的条件解析
for op in operators:
pattern = rf'(\w+)\s*{re.escape(op)}\s*([\'\"]?[^\s\'\"]+[\'\"]?)'
matches = re.findall(pattern, where_clause, re.IGNORECASE)
for match in matches:
field, value = match
conditions.append({
"field": field,
"operator": op,
"value": value.strip('\"\'')
})
return conditions
def _build_message_context(self, message: 'MessageExt') -> Dict[str, Any]:
"""构建消息上下文"""
context = {
"tags": message.tags or "",
"keys": message.keys or "",
"msgId": message.msg_id,
"topic": message.topic,
"queueId": message.queue_id,
"bornTimestamp": message.born_timestamp,
"storeTimestamp": message.store_timestamp
}
# 添加用户属性
if hasattr(message, 'properties') and message.properties:
for key, value in message.properties.items():
context[key] = value
return context
def _evaluate_expression(self, compiled_expr: Dict[str, Any],
context: Dict[str, Any]) -> bool:
"""评估表达式"""
conditions = compiled_expr.get("conditions", [])
if not conditions:
return True
# 简单的AND逻辑（实际应该支持更复杂的逻辑）
for condition in conditions:
if not self._evaluate_condition(condition, context):
return False
return True
def _evaluate_condition(self, condition: Dict[str, Any],
context: Dict[str, Any]) -> bool:
"""评估单个条件"""
field = condition["field"]
operator = condition["operator"]
expected_value = condition["value"]
actual_value = context.get(field, "")
try:
if operator == "=":
return str(actual_value) == str(expected_value)
elif operator == "!=":
return str(actual_value) != str(expected_value)
elif operator == ">":
return float(actual_value) > float(expected_value)
elif operator == "<":
return float(actual_value) < float(expected_value)
elif operator == ">=":
return float(actual_value) >= float(expected_value)
elif operator == "<=":
return float(actual_value) <= float(expected_value)
elif operator.upper() == "LIKE":
import re
pattern = expected_value.replace("%", ".*")
return bool(re.match(pattern, str(actual_value)))
elif operator.upper() == "IN":
values = expected_value.strip("()").split(",")
values = [v.strip().strip("'\"") for v in values]
return str(actual_value) in values
else:
return False
except (ValueError, TypeError):
return False
# 过滤消息生产者
class FilterMessageProducer:
"""支持过滤的消息生产者"""
def __init__(self, producer_config: 'ProducerConfig'):
self.config = producer_config
self.producer = None
self.send_stats = {
"total_sent": 0,
"success_sent": 0,
"failed_sent": 0
}
def start(self):
"""启动生产者"""
from rocketmq_producer import DefaultMQProducer
self.producer = DefaultMQProducer(self.config)
self.producer.start()
print("过滤消息生产者启动成功")
def shutdown(self):
"""关闭生产者"""
if self.producer:
self.producer.shutdown()
print("过滤消息生产者关闭成功")
def send_message_with_tags(self, topic: str, tags: str, keys: str,
body: str, properties: Dict[str, str] = None) -> 'SendResult':
"""发送带标签的消息"""
from rocketmq_producer import Message
message = Message(
topic=topic,
tags=tags,
keys=keys,
body=body.encode('utf-8')
)
# 设置用户属性
if properties:
for key, value in properties.items():
message.put_property(key, value)
try:
result = self.producer.send(message)
self.send_stats["total_sent"] += 1
self.send_stats["success_sent"] += 1
return result
except Exception as e:
self.send_stats["total_sent"] += 1
self.send_stats["failed_sent"] += 1
raise e
def send_messages_batch_with_different_tags(self, topic: str,
message_data: List[Dict[str, Any]]) -> List['SendResult']:
"""批量发送不同标签的消息"""
results = []
for data in message_data:
try:
result = self.send_message_with_tags(
topic=topic,
tags=data.get("tags", ""),
keys=data.get("keys", ""),
body=data.get("body", ""),
properties=data.get("properties", {})
)
results.append(result)
except Exception as e:
print(f"发送消息失败: {e}")
results.append(None)
return results
def get_send_stats(self) -> Dict[str, int]:
"""获取发送统计"""
return self.send_stats.copy()
# 过滤消息消费者
class FilterMessageConsumer:
"""支持过滤的消息消费者"""
def __init__(self, consumer_config: 'ConsumerConfig'):
self.config = consumer_config
self.consumer = None
self.message_filter: Optional[MessageFilter] = None
self.filter_stats = {
"total_received": 0,
"filtered_out": 0,
"processed": 0
}
def start(self):
"""启动消费者"""
from rocketmq_consumer import DefaultMQPushConsumer
self.consumer = DefaultMQPushConsumer(self.config)
# 注册过滤消息监听器
listener = FilterMessageListener(self.message_filter, self.filter_stats)
self.consumer.register_message_listener(listener)
self.consumer.start()
print("过滤消息消费者启动成功")
def shutdown(self):
"""关闭消费者"""
if self.consumer:
self.consumer.shutdown()
print("过滤消息消费者关闭成功")
def subscribe_with_tag_filter(self, topic: str, tag_expression: str):
"""使用标签过滤订阅主题"""
self.message_filter = TagFilter(tag_expression)
if self.consumer:
self.consumer.subscribe(topic, tag_expression)
def subscribe_with_sql_filter(self, topic: str, sql_expression: str):
"""使用SQL过滤订阅主题"""
self.message_filter = SQL92Filter(sql_expression)
if self.consumer:
# 注意：实际的RocketMQ需要在Broker端支持SQL过滤
self.consumer.subscribe(topic, "*") # 客户端过滤
def get_filter_stats(self) -> Dict[str, int]:
"""获取过滤统计"""
return self.filter_stats.copy()
# 过滤消息监听器
class FilterMessageListener:
"""过滤消息监听器"""
def __init__(self, message_filter: Optional[MessageFilter],
stats: Dict[str, int]):
self.message_filter = message_filter
self.stats = stats
def consume_message(self, messages: List['MessageExt'],
context: 'ConsumeContext') -> 'ConsumeResult':
"""消费消息"""
try:
filtered_messages = []
for message in messages:
self.stats["total_received"] += 1
# 应用过滤器
if self.message_filter and not self.message_filter.match(message):
self.stats["filtered_out"] += 1
continue
filtered_messages.append(message)
# 处理过滤后的消息
if filtered_messages:
result = self._process_filtered_messages(filtered_messages)
if result == ConsumeResult.SUCCESS:
self.stats["processed"] += len(filtered_messages)
return result
return ConsumeResult.SUCCESS
except Exception as e:
print(f"过滤消息处理异常: {e}")
return ConsumeResult.RECONSUME_LATER
def _process_filtered_messages(self, messages: List['MessageExt']) -> 'ConsumeResult':
"""处理过滤后的消息"""
try:
for message in messages:
print(f"处理过滤消息:")
print(f" ID: {message.msg_id}")
print(f" Topic: {message.topic}")
print(f" Tags: {message.tags}")
print(f" Keys: {message.keys}")
print(f" 内容: {message.body.decode('utf-8')[:100]}...")
# 打印用户属性
if hasattr(message, 'properties') and message.properties:
print(f" 属性: {message.properties}")
print()
return ConsumeResult.SUCCESS
except Exception as e:
print(f"处理过滤消息异常: {e}")
return ConsumeResult.RECONSUME_LATER
5.5.2 消息过滤使用示例
# 消息过滤使用示例
class MessageFilterExample:
"""消息过滤使用示例"""
def __init__(self):
# 配置生产者
self.producer_config = ProducerConfig(
producer_group="filter_producer_group",
name_server_addr="localhost:9876"
)
# 配置消费者
self.consumer_config = ConsumerConfig(
consumer_group="filter_consumer_group",
name_server_addr="localhost:9876"
)
# 创建生产者和消费者
self.producer = FilterMessageProducer(self.producer_config)
self.consumer = FilterMessageConsumer(self.consumer_config)
def start_producer_example(self):
"""启动生产者示例"""
try:
self.producer.start()
# 发送不同标签的消息
self._send_messages_with_different_tags()
# 发送带属性的消息
self._send_messages_with_properties()
# 打印发送统计
stats = self.producer.get_send_stats()
print(f"\n发送统计: {stats}")
finally:
self.producer.shutdown()
def start_tag_filter_consumer_example(self):
"""启动标签过滤消费者示例"""
try:
# 订阅特定标签的消息
self.consumer.subscribe_with_tag_filter("FilterTopic", "VIP || IMPORTANT")
# 启动消费者
self.consumer.start()
print("标签过滤消费者启动成功，等待消息...")
# 等待消息
while True:
time.sleep(10)
stats = self.consumer.get_filter_stats()
print(f"过滤统计: {stats}")
except KeyboardInterrupt:
print("收到中断信号，正在关闭消费者...")
finally:
self.consumer.shutdown()
def start_sql_filter_consumer_example(self):
"""启动SQL过滤消费者示例"""
try:
# 使用SQL表达式过滤消息
sql_expression = "WHERE level > 5 AND category = 'order'"
self.consumer.subscribe_with_sql_filter("FilterTopic", sql_expression)
# 启动消费者
self.consumer.start()
print("SQL过滤消费者启动成功，等待消息...")
# 等待消息
while True:
time.sleep(10)
stats = self.consumer.get_filter_stats()
print(f"过滤统计: {stats}")
except KeyboardInterrupt:
print("收到中断信号，正在关闭消费者...")
finally:
self.consumer.shutdown()
def _send_messages_with_different_tags(self):
"""发送不同标签的消息"""
message_data = [
{
"tags": "VIP",
"keys": "vip_user_001",
"body": "VIP用户订单消息",
"properties": {"level": "10", "category": "order"}
},
{
"tags": "NORMAL",
"keys": "normal_user_001",
"body": "普通用户订单消息",
"properties": {"level": "3", "category": "order"}
},
{
"tags": "IMPORTANT",
"keys": "important_001",
"body": "重要系统消息",
"properties": {"level": "8", "category": "system"}
},
{
"tags": "DEBUG",
"keys": "debug_001",
"body": "调试消息",
"properties": {"level": "1", "category": "debug"}
}
]
results = self.producer.send_messages_batch_with_different_tags(
"FilterTopic", message_data
)
for i, result in enumerate(results):
if result:
print(f"发送消息 {i+1} 成功: {result.msg_id}")
else:
print(f"发送消息 {i+1} 失败")
def _send_messages_with_properties(self):
"""发送带属性的消息"""
# 发送高级别订单消息
self.producer.send_message_with_tags(
topic="FilterTopic",
tags="ORDER",
keys="high_level_order",
body="高级别订单消息",
properties={
"level": "9",
"category": "order",
"priority": "high",
"amount": "1000.00"
}
)
# 发送低级别通知消息
self.producer.send_message_with_tags(
topic="FilterTopic",
tags="NOTIFICATION",
keys="low_level_notification",
body="低级别通知消息",
properties={
"level": "2",
"category": "notification",
"priority": "low"
}
)
# 过滤器测试
class FilterTest:
"""过滤器测试"""
def test_tag_filter(self):
"""测试标签过滤器"""
print("测试标签过滤器:")
# 创建测试消息
test_messages = [
self._create_test_message("VIP"),
self._create_test_message("NORMAL"),
self._create_test_message("IMPORTANT"),
self._create_test_message("DEBUG")
]
# 测试不同的标签过滤器
filters = [
TagFilter("VIP"),
TagFilter("VIP || IMPORTANT"),
TagFilter("*"),
TagFilter("PREMIUM")
]
for filter_obj in filters:
print(f"\n过滤表达式: {filter_obj.get_filter_expression()}")
for msg in test_messages:
match = filter_obj.match(msg)
print(f" 消息标签 '{msg.tags}': {'匹配' if match else '不匹配'}")
def test_sql_filter(self):
"""测试SQL过滤器"""
print("\n测试SQL过滤器:")
# 创建带属性的测试消息
test_messages = [
self._create_test_message_with_properties("ORDER", {"level": "8", "category": "order"}),
self._create_test_message_with_properties("ORDER", {"level": "3", "category": "order"}),
self._create_test_message_with_properties("NOTIFICATION", {"level": "6", "category": "notification"}),
self._create_test_message_with_properties("DEBUG", {"level": "1", "category": "debug"})
]
# 测试不同的SQL过滤器
sql_filters = [
"WHERE level > 5",
"WHERE level > 5 AND category = 'order'",
"WHERE category = 'notification'",
"WHERE level <= 3"
]
for sql_expr in sql_filters:
filter_obj = SQL92Filter(sql_expr)
print(f"\nSQL表达式: {sql_expr}")
for msg in test_messages:
match = filter_obj.match(msg)
props = getattr(msg, 'properties', {})
print(f" 消息 (level={props.get('level', 'N/A')}, category={props.get('category', 'N/A')}): {'匹配' if match else '不匹配'}")
def _create_test_message(self, tags: str) -> 'MessageExt':
"""创建测试消息"""
from rocketmq_consumer import MessageExt
return MessageExt(
topic="TestTopic",
tags=tags,
keys=f"test_key_{tags.lower()}",
body=f"测试消息内容 - {tags}".encode('utf-8'),
msg_id=f"test_msg_{tags.lower()}",
queue_id=0,
queue_offset=0,
born_timestamp=int(time.time() * 1000),
store_timestamp=int(time.time() * 1000)
)
def _create_test_message_with_properties(self, tags: str,
properties: Dict[str, str]) -> 'MessageExt':
"""创建带属性的测试消息"""
msg = self._create_test_message(tags)
msg.properties = properties
return msg
# 使用示例
if __name__ == "__main__":
# 测试过滤器
filter_test = FilterTest()
filter_test.test_tag_filter()
filter_test.test_sql_filter()
# 消息过滤示例
example = MessageFilterExample()
# 发送消息示例
example.start_producer_example()
# 标签过滤消费示例
# example.start_tag_filter_consumer_example()
# SQL过滤消费示例
# example.start_sql_filter_consumer_example()
5.6 本章总结
5.6.1 核心知识点
# RocketMQ高级特性总结
class RocketMQAdvancedFeaturesSummary:
"""RocketMQ高级特性总结"""
def __init__(self):
self.features = {
"顺序消息": {
"描述": "保证消息按照发送顺序被消费",
"类型": ["全局顺序", "分区顺序"],
"关键点": [
"使用MessageQueueSelector选择队列",
"同一队列内消息顺序消费",
"消费者需要顺序处理消息"
],
"适用场景": [
"订单状态变更",
"账户余额变动",
"库存变化记录"
]
},
"事务消息": {
"描述": "保证本地事务与消息发送的一致性",
"流程": [
"发送半消息",
"执行本地事务",
"提交或回滚事务消息"
],
"关键点": [
"实现TransactionListener接口",
"处理事务状态检查",
"确保幂等性"
],
"适用场景": [
"分布式事务",
"数据一致性保证",
"业务解耦"
]
},
"延时消息": {
"描述": "消息在指定时间后才能被消费",
"实现方式": [
"预定义延时级别",
"指定延时时间"
],
"关键点": [
"支持18个延时级别",
"最大延时2小时",
"延时精度秒级"
],
"适用场景": [
"订单超时取消",
"定时任务",
"延时通知"
]
},
"批量消息": {
"描述": "一次发送多条消息，提高吞吐量",
"优势": [
"减少网络开销",
"提高发送效率",
"降低系统负载"
],
"限制": [
"消息总大小不超过4MB",
"同一批次消息必须是同一Topic",
"不支持延时消息"
],
"适用场景": [
"日志收集",
"数据同步",
"批量通知"
]
},
"消息过滤": {
"描述": "根据条件过滤消息，减少无效消费",
"类型": [
"标签过滤（Tag Filter）",
"SQL92过滤（SQL Filter）"
],
"关键点": [
"支持多标签过滤",
"SQL表达式过滤",
"客户端和服务端过滤"
],
"适用场景": [
"消息分类处理",
"条件消费",
"减少网络传输"
]
}
}
def get_feature_summary(self, feature_name: str) -> Dict[str, Any]:
"""获取特性总结"""
return self.features.get(feature_name, {})
def get_all_features(self) -> List[str]:
"""获取所有特性列表"""
return list(self.features.keys())
def print_summary(self):
"""打印特性总结"""
print("RocketMQ高级特性总结:")
print("=" * 50)
for feature_name, feature_info in self.features.items():
print(f"\n{feature_name}:")
print(f" 描述: {feature_info['描述']}")
for key, value in feature_info.items():
if key != '描述':
print(f" {key}:")
if isinstance(value, list):
for item in value:
print(f" - {item}")
else:
print(f" {value}")
5.6.2 最佳实践
# RocketMQ高级特性最佳实践
class RocketMQAdvancedBestPractices:
"""RocketMQ高级特性最佳实践"""
def __init__(self):
self.best_practices = {
"顺序消息最佳实践": [
"合理选择分区键，避免热点队列",
"消费者处理失败时要谨慎重试",
"避免长时间阻塞消费线程",
"监控队列消费进度",
"考虑消费者扩容对顺序的影响"
],
"事务消息最佳实践": [
"本地事务要保证幂等性",
"事务状态检查要快速响应",
"合理设置事务超时时间",
"记录事务执行日志",
"处理事务回查逻辑"
],
"延时消息最佳实践": [
"选择合适的延时级别",
"避免大量相同时间的延时消息",
"考虑时钟偏移问题",
"监控延时消息堆积",
"设计延时消息的取消机制"
],
"批量消息最佳实践": [
"控制批量大小，避免超过限制",
"合理设置批量发送间隔",
"处理批量发送失败的重试",
"监控批量发送性能",
"考虑消息顺序要求"
],
"消息过滤最佳实践": [
"优先使用标签过滤",
"SQL过滤表达式要简洁",
"避免复杂的过滤逻辑",
"在生产者端设置合适的标签",
"监控过滤效果和性能"
],
"性能优化实践": [
"合理配置生产者和消费者参数",
"使用异步发送提高吞吐量",
"批量处理消息",
"避免频繁创建连接",
"监控关键性能指标"
],
"可靠性保证实践": [
"实现消息幂等处理",
"设计合理的重试机制",
"处理死信队列消息",
"监控消息堆积情况",
"建立消息追踪机制"
]
}
def get_best_practices(self, category: str) -> List[str]:
"""获取最佳实践"""
return self.best_practices.get(category, [])
def print_best_practices(self):
"""打印最佳实践"""
print("RocketMQ高级特性最佳实践:")
print("=" * 50)
for category, practices in self.best_practices.items():
print(f"\n{category}:")
for i, practice in enumerate(practices, 1):
print(f" {i}. {practice}")
# 性能监控和调优
class RocketMQPerformanceMonitor:
"""RocketMQ性能监控"""
def __init__(self):
self.metrics = {
"生产者指标": [
"发送TPS（每秒事务数）",
"发送延迟（RT）",
"发送成功率",
"发送失败率",
"重试次数"
],
"消费者指标": [
"消费TPS",
"消费延迟",
"消费成功率",
"消费失败率",
"消息堆积量"
],
"Broker指标": [
"消息存储量",
"磁盘使用率",
"内存使用率",
"网络IO",
"队列深度"
],
"系统指标": [
"CPU使用率",
"内存使用率",
"磁盘IO",
"网络带宽",
"连接数"
]
}
def get_monitoring_metrics(self) -> Dict[str, List[str]]:
"""获取监控指标"""
return self.metrics
def print_monitoring_guide(self):
"""打印监控指南"""
print("RocketMQ性能监控指南:")
print("=" * 50)
for category, metrics in self.metrics.items():
print(f"\n{category}:")
for metric in metrics:
print(f" - {metric}")
print("\n监控建议:")
print(" 1. 建立完善的监控体系")
print(" 2. 设置合理的告警阈值")
print(" 3. 定期分析性能趋势")
print(" 4. 建立性能基线")
print(" 5. 制定性能调优计划")
5.6.3 练习题
# 练习题
class RocketMQAdvancedExercises:
"""RocketMQ高级特性练习题"""
def __init__(self):
self.exercises = [
{
"题目": "实现一个电商订单处理系统",
"要求": [
"使用顺序消息保证订单状态变更的顺序",
"使用事务消息保证库存扣减和订单创建的一致性",
"使用延时消息实现订单超时自动取消",
"使用消息过滤实现不同类型订单的分类处理"
],
"提示": [
"以订单ID作为分区键保证顺序",
"在事务监听器中处理库存操作",
"设置合适的订单超时时间",
"使用标签区分订单类型"
]
},
{
"题目": "设计一个日志收集系统",
"要求": [
"使用批量消息提高日志发送效率",
"使用消息过滤实现不同级别日志的处理",
"实现日志消息的压缩和解压缩",
"监控日志发送和消费的性能指标"
],
"提示": [
"按时间或数量触发批量发送",
"使用日志级别作为过滤条件",
"在消息体中实现压缩逻辑",
"记录发送和消费的统计信息"
]
},
{
"题目": "实现一个分布式任务调度系统",
"要求": [
"使用延时消息实现定时任务",
"使用事务消息保证任务状态的一致性",
"使用消息过滤实现任务类型的分发",
"实现任务的重试和失败处理机制"
],
"提示": [
"根据执行时间计算延时级别",
"在事务中更新任务状态",
"使用任务类型作为标签",
"设计合理的重试策略"
]
},
{
"题目": "优化现有的消息系统性能",
"要求": [
"分析当前系统的性能瓶颈",
"使用批量消息优化发送性能",
"使用异步发送提高吞吐量",
"实现消息的监控和告警"
],
"提示": [
"监控发送和消费的延迟",
"合理设置批量大小和间隔",
"使用回调处理异步结果",
"建立完善的监控体系"
]
}
]
def get_exercise(self, index: int) -> Dict[str, Any]:
"""获取练习题"""
if 0 <= index < len(self.exercises):
return self.exercises[index]
return {}
def print_all_exercises(self):
"""打印所有练习题"""
print("RocketMQ高级特性练习题:")
print("=" * 50)
for i, exercise in enumerate(self.exercises, 1):
print(f"\n练习题 {i}: {exercise['题目']}")
print("要求:")
for req in exercise['要求']:
print(f" - {req}")
print("提示:")
for hint in exercise['提示']:
print(f" - {hint}")
print()
# 使用示例
if __name__ == "__main__":
# 打印特性总结
summary = RocketMQAdvancedFeaturesSummary()
summary.print_summary()
print("\n" + "=" * 80 + "\n")
# 打印最佳实践
best_practices = RocketMQAdvancedBestPractices()
best_practices.print_best_practices()
print("\n" + "=" * 80 + "\n")
# 打印监控指南
monitor = RocketMQPerformanceMonitor()
monitor.print_monitoring_guide()
print("\n" + "=" * 80 + "\n")
# 打印练习题
exercises = RocketMQAdvancedExercises()
exercises.print_all_exercises()
本章小结：
本章详细介绍了RocketMQ的五大高级特性：
- 顺序消息 - 保证消息的有序处理，适用于对顺序有严格要求的业务场景
- 事务消息 - 保证分布式事务的一致性，解决本地事务与消息发送的原子性问题
- 延时消息 - 实现消息的延时投递，支持定时任务和延时处理场景
- 批量消息 - 提高消息发送效率，减少网络开销，适用于高吞吐量场景
- 消息过滤 - 实现消息的精确投递，减少无效消费，提高系统效率
通过学习这些高级特性，你可以根据不同的业务需求选择合适的消息处理方式，构建更加高效、可靠的分布式消息系统。在实际应用中，要结合业务特点和性能要求，合理使用这些特性，并建立完善的监控和运维体系。
文章版权声明：除非注明，否则均为边学边练网络文章，版权归原作者所有
