| 提交 | 用户 | 时间 | ||
| e7c126 | 1 | package com.iailab.framework.tenant.core.mq.kafka; |
| H | 2 | |
| 3 | import cn.hutool.core.util.ReflectUtil; | |
| 4 | import com.iailab.framework.tenant.core.context.TenantContextHolder; | |
| 5 | import org.apache.kafka.clients.producer.ProducerInterceptor; | |
| 6 | import org.apache.kafka.clients.producer.ProducerRecord; | |
| 7 | import org.apache.kafka.clients.producer.RecordMetadata; | |
| 8 | import org.apache.kafka.common.header.Headers; | |
| 9 | import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; | |
| 10 | ||
| 11 | import java.util.Map; | |
| 12 | ||
| 13 | import static com.iailab.framework.web.core.util.WebFrameworkUtils.HEADER_TENANT_ID; | |
| 14 | ||
| 15 | /** | |
| 16 | * Kafka 消息队列的多租户 {@link ProducerInterceptor} 实现类 | |
| 17 | * | |
| 18 | * 1. Producer 发送消息时，将 {@link TenantContextHolder} 租户编号，添加到消息的 Header 中 | |
| 19 | * 2. Consumer 消费消息时，将消息的 Header 的租户编号，添加到 {@link TenantContextHolder} 中，通过 {@link InvocableHandlerMethod} 实现 | |
| 20 | * | |
| 21 | * @author iailab | |
| 22 | */ | |
| 23 | public class TenantKafkaProducerInterceptor implements ProducerInterceptor<Object, Object> { | |
| 24 | ||
| 25 | @Override | |
| 26 | public ProducerRecord<Object, Object> onSend(ProducerRecord<Object, Object> record) { | |
| 27 | Long tenantId = TenantContextHolder.getTenantId(); | |
| 28 | if (tenantId != null) { | |
| 29 | Headers headers = (Headers) ReflectUtil.getFieldValue(record, "headers"); // private 属性，没有 get 方法，智能反射 | |
| 30 | headers.add(HEADER_TENANT_ID, tenantId.toString().getBytes()); | |
| 31 | } | |
| 32 | return record; | |
| 33 | } | |
| 34 | ||
| 35 | @Override | |
| 36 | public void onAcknowledgement(RecordMetadata metadata, Exception exception) { | |
| 37 | } | |
| 38 | ||
| 39 | @Override | |
| 40 | public void close() { | |
| 41 | } | |
| 42 | ||
| 43 | @Override | |
| 44 | public void configure(Map<String, ?> configs) { | |
| 45 | } | |
| 46 | ||
| 47 | } | |
