spring-boot-restapi和spring-kafka

7lrncoxx 于 2021-06-04 发布在 Kafka

关注(0)|答案(1)|浏览(425)

我设计了一个springbootsrestapi添加和获取方法

@RestController("ProductV1Controller")
    public class ProductController 
     {

         private final IProductProducer _productProducer;
         public ProductController(IProductProducer productProducer) {
        _productProducer = productProducer;}

         @PostMapping()
            void AddProduct(@Valid @RequestBody ProductViewModel product) {
                _productProducer.AddProduct(product);
            }

        @GetMapping()
            List<ProductViewModel> Products() {
                var test = _productProducer.GetProducts();
                return _productProducer.GetProducts();
            }
}

服务层

@Service

    public class ProductProducer implements IProductProducer{
        private final KafkaTemplate<String, Object> _template;

        public ProductProducer(KafkaTemplate<String, Object> _template) {
            this._template = _template;
        }

        @Override
        public List<ProductViewModel> GetProducts() {
            this._template.send(ProductTopicConstants.GET_PRODUCTS,null);
            return List.of(new ProductViewModel("","",0,"")); --> Need to return the value from the kafka
        }

        @Override
        public void AddProduct(ProductViewModel product) {
            this._template.send(ProductTopicConstants.ADD_PRODUCT, product);
        }

    }

Kafka听众

@KafkaListener(id = ProductTopicConstants.GET_PRODUCTS, topics = ProductTopicConstants.GET_PRODUCTS)
    public List<Product> GetProducts() {
        return _productRepository.findAll();
    }

在服务层 GetProducts() 我需要返回来自 _productRepository.findAll(); 使用springkafka实现restapi的最佳方法是什么。

Java spring apache-kafka spring-boot spring-kafka

来源：https://stackoverflow.com/questions/63058608/spring-boot-rest-api-with-spring-kafka

1条答案

按热度按时间

你需要使用 ReplyingKafkaTemplate 将结果返回给rest控制器。
请参阅回复kafkatemplate。
版本2.1.3引入了kafkatemplate的子类来提供请求/应答语义。这个类名为replyingkafkatemplate，并且有一个方法（除了超类中的方法之外）。
结果是一个listenablefuture，异步填充结果（或者超时时的异常）。结果还有一个sendfuture属性，它是调用kafkatemplate.send（）的结果。您可以使用这个未来值来确定发送操作的结果。
文档中有一个示例。
编辑

@SpringBootApplication
@RestController
public class So63058608Application {

    private static final Logger LOG = LoggerFactory.getLogger(So63058608Application.class);

    public static void main(String[] args) {
        SpringApplication.run(So63058608Application.class, args);
    }

    @Autowired
    private ReplyingKafkaTemplate<String, String, List<String>> replyTemplate;

    @GetMapping(path = "/get")
    public List<String> getThem() throws Exception {
        RequestReplyFuture<String, String, List<String>> future =
                this.replyTemplate.sendAndReceive(new ProducerRecord<>("so63058608-1", 0, null, null));
        LOG.info(future.getSendFuture().get(10, TimeUnit.SECONDS).getRecordMetadata().toString());
        return future.get(10, TimeUnit.SECONDS).value();
    }

    @KafkaListener(id = "so63058608-1", topics = "so63058608-1", splitIterables = false)
    @SendTo
    public List<String> returnList(@Payload(required = false) String payload) {
        return new ArrayList<>(List.of("foo", "bar", "baz"));
    }

    @Bean
    public ReplyingKafkaTemplate<String, String, List<String>> replyer(ProducerFactory<String, String> pf,
            ConcurrentKafkaListenerContainerFactory<String, List<String>> containerFactory) {

        containerFactory.setReplyTemplate(kafkaTemplate(pf));
        ConcurrentMessageListenerContainer<String, List<String>> container = replyContainer(containerFactory);
        ReplyingKafkaTemplate<String, String, List<String>> replyer = new ReplyingKafkaTemplate<>(pf, container);
        return replyer;
    }

    @Bean
    public ConcurrentMessageListenerContainer<String, List<String>> replyContainer(
            ConcurrentKafkaListenerContainerFactory<String, List<String>> containerFactory) {

        ConcurrentMessageListenerContainer<String, List<String>> container =
                containerFactory.createContainer("so63058608-2");
        container.getContainerProperties().setGroupId("so63058608-2");
        container.setBatchErrorHandler(new BatchLoggingErrorHandler());
        return container;
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> pf) {
        return new KafkaTemplate<>(pf);
    }

    @Bean
    public NewTopic topic1() {
        return TopicBuilder.name("so63058608-1").partitions(1).replicas(1).build();
    }

    @Bean
    public NewTopic topic3() {
        return TopicBuilder.name("so63058608-2").partitions(1).replicas(1).build();
    }

}

spring.kafka.consumer.key-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.properties.spring.json.trusted.packages=*

spring.kafka.producer.key-serializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

$ curl localhost:8080/get
["foo","bar","baz"]

编辑2
返回了一个对象列表。。。

@SpringBootApplication
@RestController
public class So63058608Application {

    private static final Logger LOG = LoggerFactory.getLogger(So63058608Application.class);

    public static void main(String[] args) {
        SpringApplication.run(So63058608Application.class, args);
    }

    @Autowired
    private ReplyingKafkaTemplate<String, String, List<Foo>> replyTemplate;

    @GetMapping(path = "/get")
    public List<Foo> getThem() throws Exception {
        RequestReplyFuture<String, String, List<Foo>> future =
                this.replyTemplate.sendAndReceive(new ProducerRecord<>("so63058608-1", 0, null, null));
        LOG.info(future.getSendFuture().get(10, TimeUnit.SECONDS).getRecordMetadata().toString());
        List<Foo> result = future.get(10, TimeUnit.SECONDS).value();
        LOG.info(result.toString());
        return result;
    }

    @KafkaListener(id = "so63058608-1", topics = "so63058608-1", splitIterables = false)
    @SendTo
    public List<Foo> returnList(@Payload(required = false) String payload) {
        return new ArrayList<>(List.of(new Foo("foo"), new Foo("bar"), new Foo("baz")));
    }

    @Bean
    public ReplyingKafkaTemplate<String, String, List<Foo>> replyer(ProducerFactory<String, String> pf,
            ConcurrentKafkaListenerContainerFactory<String, List<Foo>> containerFactory) {

        containerFactory.setReplyTemplate(kafkaTemplate(pf));
        ConcurrentMessageListenerContainer<String, List<Foo>> container = replyContainer(containerFactory);
        ReplyingKafkaTemplate<String, String, List<Foo>> replyer = new ReplyingKafkaTemplate<>(pf, container);
        return replyer;
    }

    @Bean
    public ConcurrentMessageListenerContainer<String, List<Foo>> replyContainer(
            ConcurrentKafkaListenerContainerFactory<String, List<Foo>> containerFactory) {

        ConcurrentMessageListenerContainer<String, List<Foo>> container =
                containerFactory.createContainer("so63058608-2");
        container.getContainerProperties().setGroupId("so63058608-2");
        container.setBatchErrorHandler(new BatchLoggingErrorHandler());
        return container;
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> pf) {
        return new KafkaTemplate<>(pf);
    }

    @Bean
    public NewTopic topic1() {
        return TopicBuilder.name("so63058608-1").partitions(1).replicas(1).build();
    }

    @Bean
    public NewTopic topic3() {
        return TopicBuilder.name("so63058608-2").partitions(1).replicas(1).build();
    }

    public static JavaType returnType(byte[] data, Headers headers) {
        return TypeFactory.defaultInstance()
                .constructCollectionLikeType(List.class, Foo.class);
    }

}

class Foo {

    private String bar;

    public Foo() {
    }

    public Foo(String bar) {
        this.bar = bar;
    }

    public String getBar() {
        return this.bar;
    }

    public void setBar(String bar) {
        this.bar = bar;
    }

    @Override
    public String toString() {
        return "Foo [bar=" + this.bar + "]";
    }

}

spring.kafka.consumer.properties.spring.json.value.type.method=com.example.demo.So63058608Application.returnType

[Foo [bar=foo], Foo [bar=bar], Foo [bar=baz]]

赞(0）回复(0）举报 2021-06-04

相关问题

热门标签

Java query python Node 开发语言 request Util 数据库 Table 后端算法 Logger Message Element Parser

最新问答

xxl-job 安全组扫描到执行器端口服务存在信息泄露漏洞
回答(1) 发布于 4个月前
xxl-job 不能和nacos兼容？
回答(3) 发布于 4个月前
xxl-job 任务执行完后无法结束，日志一直转圈
回答(3) 发布于 4个月前
xxl-job-admin页面上查看调度日志样式问题
回答(1) 发布于 4个月前
xxl-job 参数512字符限制能否去掉
回答(1) 发布于 4个月前