Java
KafkaRequestExecutor
Kafka RPC 请求执行器，实现请求-响应模式的消息消费。
- Type:
final class(package-private) - Package:
com.euonia.bus - Extends:
KafkaRecipient - Implements:
Executor
Fields¶
| Name | Type | Description |
|---|---|---|
consumer |
KafkaConsumer<String, byte[]> |
Kafka 消费者实例 |
replyProducer |
KafkaProducer<String, byte[]> |
回复消息的生产者实例 |
Methods¶
| Name | Signature | Description |
|---|---|---|
KafkaRequestExecutor |
KafkaRequestExecutor(KafkaBusOptions options, HandlerContext handler, MessageSerializer serializer, Class<?> messageType) |
构造器 |
start |
void start(String channelName) |
启动执行器，订阅 topic 并轮询消息；收到消息后异步调用 handler，完成后发送回复并触发确认事件 |
createConsumer |
KafkaConsumer<String, byte[]> createConsumer(String channelName) |
根据 options 创建 Kafka 消费者 |
close |
void close() |
关闭消费者和回复生产者 |
createReplyProducer |
KafkaProducer<String, byte[]> createReplyProducer() |
创建回复消息的生产者 |
sendReply |
void sendReply(KafkaProducer<String, byte[]> producer, ConsumerRecord<String, byte[]> request, MessageEnvelope<?> message, Object result, Throwable error) |
发送回复消息到指定的 replyTo topic；根据处理结果（异常/消息/void）设置不同的类型标记 |