在一个kafka主题下发送两个序列化的java对象

zbq4xfa0  于 2021-07-24  发布在  Java
关注(0)|答案(2)|浏览(472)

我想实现kafka消费者和生产者发送和接收java对象。完整来源我试过这个：
制作人：

  1. @Configuration
  2. public class KafkaProducerConfig {
  4.     @Value(value = "${kafka.bootstrapAddress}")
  5.     private String bootstrapAddress;
  7.     @Bean
  8.     public ProducerFactory<String, SaleRequestFactory> saleRequestFactoryProducerFactory() {
  9.         Map<String, Object> configProps = new HashMap<>();
  10.         configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
  11.         configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
  12.         configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, SaleRequestFactorySerializer.class);
  13.         return new DefaultKafkaProducerFactory<>(configProps);
  14.     }
  16.     @Bean
  17.     public ProducerFactory<String, String> producerFactory() {
  18.         Map<String, Object> configProps = new HashMap<>();
  19.         configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
  20.         return new DefaultKafkaProducerFactory<>(configProps);
  21.     }
  23.     @Bean
  24.     public KafkaTemplate<String, SaleRequestFactory> saleRequestFactoryKafkaTemplate() {
  25.         return new KafkaTemplate<>(saleRequestFactoryProducerFactory());
  26.     }
  28.     @Bean
  29.     public KafkaTemplate<String, String> kafkaTemplate() {
  30.         return new KafkaTemplate<>(producerFactory());
  31.     }
  33.     @Bean
  34.     public ReplyingKafkaTemplate<String, SaleRequestFactory, SaleResponseFactory> replyKafkaTemplate(ProducerFactory<String, SaleRequestFactory> producerFactory, ConcurrentKafkaListenerContainerFactory<String, SaleResponseFactory> factory) {
  35.         ConcurrentMessageListenerContainer<String, SaleResponseFactory> kafkaMessageListenerContainer = factory.createContainer("tp-sale");
  36.         kafkaMessageListenerContainer.getContainerProperties().setGroupId("tp-sale.reply");
  37.         return new ReplyingKafkaTemplate<>(producerFactory, kafkaMessageListenerContainer);
  38.     }
  39. }

发送对象：

  1. @RestController
  2. @RequestMapping("/checkout")
  3. public class CheckoutController {
  5.     private TransactionService transactionService;
  6.     private KafkaTemplate<String, SaleRequestFactory> saleRequestFactoryKafkaTemplate;
  7.     private ReplyingKafkaTemplate<String, SaleRequestFactory, SaleResponseFactory> requestReplyKafkaTemplate;
  8.     private static String topic = "tp-sale";
  10.     @Autowired
  11.     public CheckoutController(ValidationMessage validationMessage, TransactionService transactionService,
  12.                               KafkaTemplate<String, SaleRequestFactory> saleRequestFactoryKafkaTemplate,
  13.                               ReplyingKafkaTemplate<String, SaleRequestFactory, SaleResponseFactory> requestReplyKafkaTemplate){
  14.         this.transactionService = transactionService;
  15.         this.saleRequestFactoryKafkaTemplate = saleRequestFactoryKafkaTemplate;
  16.         this.requestReplyKafkaTemplate = requestReplyKafkaTemplate;
  17.     }
  19.     @PostMapping("test")
  20.     private void performPayment() throws ExecutionException, InterruptedException, TimeoutException {
  22.         Transaction transaction = new Transaction();
  23.         transaction.setStatus(PaymentTransactionStatus.IN_PROGRESS.getText());
  25.         Transaction insertedTransaction = transactionService.save(transaction);
  27.         SaleRequestFactory obj = new SaleRequestFactory();
  28.         obj.setId(100);
  30.         ProducerRecord<String, SaleRequestFactory> record = new ProducerRecord<>("tp-sale", obj);
  31.         RequestReplyFuture<String, SaleRequestFactory, SaleResponseFactory> replyFuture = requestReplyKafkaTemplate.sendAndReceive(record);
  32.         SendResult<String, SaleRequestFactory> sendResult = replyFuture.getSendFuture().get(10, TimeUnit.SECONDS);
  33.         ConsumerRecord<String, SaleResponseFactory> consumerRecord = replyFuture.get(10, TimeUnit.SECONDS);
  35.         SaleResponseFactory value = consumerRecord.value();
  36.         System.out.println("!!!!!!!!!!!! " + value.getUnique_id());
  37.     }
  38. }

消费者：

  1. @EnableKafka
  2. @Configuration
  3. public class KafkaConsumerConfig {
  5.     @Value(value = "${kafka.bootstrapAddress}")
  6.     private String bootstrapAddress;
  8.     private String groupId = "test";
  10.     @Bean
  11.     public ConsumerFactory<String, SaleResponseFactory> consumerFactory() {
  12.         Map<String, Object> props = new HashMap<>();
  13.         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
  14.         props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
  15.         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  16.         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, SaleResponseFactoryDeserializer.class);
  17.         return new DefaultKafkaConsumerFactory<>(props);
  18.     }
  20.     @Bean
  21.     public ConcurrentKafkaListenerContainerFactory<String, SaleResponseFactory> kafkaListenerContainerFactory() {
  22.         ConcurrentKafkaListenerContainerFactory<String, SaleResponseFactory> factory =
  23.                 new ConcurrentKafkaListenerContainerFactory<>();
  24.         factory.setConsumerFactory(consumerFactory());
  25.         return factory;
  26.     }
  27. }

接收对象

  1. @Component
  2. public class ProcessingSaleListener {
  4.     private static String topic = "tp-sale";
  6.     @KafkaListener(topics = "tp-sale")
  7.     public SaleResponseFactory process(@Payload SaleRequestFactory tf, @Headers MessageHeaders headers) throws Exception {
  9.         System.out.println(tf.getId());
  11.         SaleResponseFactory resObj = new SaleResponseFactory();
  12.         resObj.setUnique_id("123123");
  14.         return resObj;
  15.     }
  16. }

自定义对象

  1. import java.io.Serializable;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Builder;
  5. import lombok.Getter;
  6. import lombok.NoArgsConstructor;
  7. import lombok.Setter;
  9. @Getter
  10. @Setter
  11. @NoArgsConstructor
  12. @AllArgsConstructor
  13. @Builder(toBuilder = true)
  14. public class SaleRequestFactory implements Serializable {
  16.     private static final long serialVersionUID = 1744050117179344127L;
  18.     private int id;
  20. }

序列化程序

  1. import org.apache.kafka.common.serialization.Serializer;
  3. import java.io.ByteArrayOutputStream;
  4. import java.io.IOException;
  5. import java.io.ObjectOutputStream;
  6. import java.io.Serializable;
  8. public class SaleRequestFactorySerializer implements Serializable, Serializer<SaleRequestFactory> {
  10.     @Override
  11.     public byte[] serialize(String topic, SaleRequestFactory data)
  12.     {
  13.         ByteArrayOutputStream out = new ByteArrayOutputStream();
  14.         try
  15.         {
  16.             ObjectOutputStream outputStream = new ObjectOutputStream(out);
  17.             outputStream.writeObject(data);
  18.             out.close();
  19.         }
  20.         catch (IOException e)
  21.         {
  22.             throw new RuntimeException("Unhandled", e);
  23.         }
  24.         return out.toByteArray();
  25.     }
  26. }

响应对象

  1. import java.io.Serializable;
  2. import java.time.LocalDateTime;
  4. import lombok.AllArgsConstructor;
  5. import lombok.Builder;
  6. import lombok.Getter;
  7. import lombok.NoArgsConstructor;
  8. import lombok.Setter;
  10. @Getter
  11. @Setter
  12. @NoArgsConstructor
  13. @AllArgsConstructor
  14. @Builder(toBuilder = true)
  15. public class SaleResponseFactory implements Serializable {
  17.     private static final long serialVersionUID = 1744050117179344127L;
  19.     private String unique_id;
  20. }

响应类

  1. import org.apache.kafka.common.serialization.Deserializer;
  3. import java.io.ByteArrayInputStream;
  4. import java.io.IOException;
  5. import java.io.ObjectInputStream;
  6. import java.io.Serializable;
  8. public class SaleResponseFactoryDeserializer implements Serializable, Deserializer<SaleRequestFactory> {
  10.     @Override
  11.     public SaleRequestFactory deserialize(String topic, byte[] data)
  12.     {
  13.         SaleRequestFactory saleRequestFactory = null;
  14.         try
  15.         {
  16.             ByteArrayInputStream bis = new ByteArrayInputStream(data);
  17.             ObjectInputStream in = new ObjectInputStream(bis);
  18.             saleRequestFactory = (SaleRequestFactory) in.readObject();
  19.             in.close();
  20.         }
  21.         catch (IOException | ClassNotFoundException e)
  22.         {
  23.             throw new RuntimeException("Unhandled", e);
  24.         }
  25.         return saleRequestFactory;
  26.     }
  27. }

我想根据对象类型发送和接收不同的序列化java对象。例如有时 SaleRequestFactory 接受 SaleResponseFactory 或发送 AuthRequestFactory 接受 AuthResponseFactory . 是否可以使用一个主题发送和接收不同的java对象？
完整示例代码

Javaspringapache-kafkaspring-bootspring-kafka

2条答案

按热度按时间
rur96b6h

rur96b6h1#

使用 Object 作为值类型-下面是一个使用boot的自动配置基础结构bean的示例。。。

  1. @SpringBootApplication
  2. public class So65866763Application {
  4.     public static void main(String[] args) {
  5.         SpringApplication.run(So65866763Application.class, args);
  6.     }
  8.     @Bean
  9.     public ApplicationRunner runner(KafkaTemplate<String, Object> template) {
  10.         return args -> {
  11.             template.send("so65866763", new Foo());
  12.             template.send("so65866763", new Bar());
  13.         };
  14.     }
  16.     @Bean
  17.     public NewTopic topic() {
  18.         return TopicBuilder.name("so65866763").partitions(1).replicas(1).build();
  19.     }
  21. }
  23. class Foo implements Serializable {
  25. }
  27. class Bar implements Serializable {
  29. }
  31. @Component
  32. @KafkaListener(id = "so65866763", topics = "so65866763")
  33. class Listener {
  35.     @KafkaHandler
  36.     void fooListener(Foo foo) {
  37.         System.out.println("In fooListener: " + foo);
  38.     }
  40.     @KafkaHandler
  41.     void barListener(Bar bar) {
  42.         System.out.println("In barListener: " + bar);
  43.     }
  45. }
  1. public class JavaSerializer implements Serializer<Object> {
  3.     @Override
  4.     public byte[] serialize(String topic, Object data) {
  5.         return null;
  6.     }
  8.     @Override
  9.     public byte[] serialize(String topic, Headers headers, Object data) {
  10.         ByteArrayOutputStream baos = new ByteArrayOutputStream();
  11.         try (ObjectOutputStream oos = new ObjectOutputStream(baos)) {
  12.             oos.writeObject(data);
  13.             return baos.toByteArray();
  14.         }
  15.         catch (IOException e) {
  16.             throw new UncheckedIOException(e);
  17.         }
  18.     }
  20. }
  1. public class JavaDeserializer implements Deserializer<Object> {
  3.     @Override
  4.     public Object deserialize(String topic, byte[] data) {
  5.         return null;
  6.     }
  8.     @Override
  9.     public Object deserialize(String topic, Headers headers, byte[] data) {
  10.         ByteArrayInputStream bais = new ByteArrayInputStream(data);
  11.         try (ObjectInputStream ois = new ObjectInputStream(bais)) {
  12.             return ois.readObject();
  13.         }
  14.         catch (IOException e) {
  15.             throw new UncheckedIOException(e);
  16.         }
  17.         catch (ClassNotFoundException e) {
  18.             throw new IllegalStateException(e);
  19.         }
  20.     }
  22. }
  1. spring.kafka.consumer.auto-offset-reset=earliest
  3. spring.kafka.producer.value-serializer=com.example.demo.JavaSerializer
  4. spring.kafka.consumer.value-deserializer=com.example.demo.JavaDeserializer
  1. In fooListener: com.example.demo.Foo@331ca660
  2. In barListener: com.example.demo.Bar@26f54288
展开查看全部
赞(0）分享  回复(0）举报  2021-07-24
uujelgoq

uujelgoq2#

这是可能的，但是每个对象类型需要两个独立的生产者工厂。或者使用bytearrayserializer并自己序列化对象（相当于gary的答案）
如果您真的想要正确地反序列化对象，那么对于使用者也是如此。否则，您将使用bytearraydeserializer（同样，相当于gary显示的反序列化程序），然后假设java无法确定字节中的对象类型（哪些序列化的对象流是}），您将在记录中包含额外的元数据，例如头，或者解析一个特定的键来确定如何反序列化数据，然后自己调用相应的反序列化方法
总的来说，我建议您重新评估为什么需要在一个主题中放置不同类型的记录，或者查看其他消息格式，包括cloudevents规范，或者使用avro/protobuf/polymorphic json类型，这些类型更适合与kafka以外的客户机一起使用

赞(0）分享  回复(0）举报  2021-07-24
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com