Spring Cloud Stream：@StreamListener处理两次消息
Spring Cloud Stream是一个用于构建消息驱动微服务的框架。它基于Spring Boot和Spring Integration，提供了一种简单且灵活的方式来处理消息。
@StreamListener是Spring Cloud Stream提供的注解，用于定义消息监听器。通过在方法上添加@StreamListener注解，可以将方法与消息队列进行绑定，当消息到达队列时，方法会被自动调用。
关于处理两次消息的问题，可能是由于消息消费失败或者重试机制导致的。在消息队列中，为了确保消息的可靠性传输，通常会有消息重试机制。当消息消费失败时，消息队列会自动进行重试，可能会导致消息被处理多次。
为了解决这个问题，可以在方法上添加一些幂等性的处理逻辑，确保多次处理不会产生副作用。例如，可以使用唯一标识符来判断消息是否已经被处理过，如果已经处理过，则直接忽略该消息。
在腾讯云中，可以使用腾讯云消息队列CMQ来实现消息驱动的微服务架构。CMQ提供了高可靠、高可用的消息队列服务，支持消息的发布和订阅，以及消息的重试机制。可以通过腾讯云消息队列CMQ的官方文档了解更多信息：腾讯云消息队列CMQ。
同时，腾讯云还提供了一系列与消息队列相关的产品和服务，如腾讯云云函数SCF、腾讯云API网关等，可以根据具体需求选择适合的产品来构建完善的消息驱动微服务架构。
相关·内容
1回答
在Spring应用程序中，我使用spring流为PubSub (spring gcp-pubsub binder)订阅一个主题。
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupI
浏览 3提问于2020-07-03得票数 1
通过下面的配置和场景，当到达maxAttempts时会发生什么？
具有Kafka绑定和以下属性的：
spring.cloud.stream.bindings.input.consumer.maxAttempts=3
spring.cloud.stream.kafka.bindings.input.consumer.autoCommitOffset=true
spring.cloud.stream.kafka.bindings.input.consumer.autoCommitOnError=false
spring.cloud.stream.kafka.bindings.inp
浏览 4提问于2017-07-13得票数 1
回答已采纳
我们有一个需求，当我们使用来自一个主题的消息时，会有一些充实发生，然后我们将消息发布到另一个主题。下面是一些事件
消费者-消费messageEnrichment -丰富了已消费的messageProducer -发布了其他主题的丰富信息。
我正在使用活页夹，一切都很好。最近，我们引入了幂等生成器并包含了transactionIdPrefix属性，并且我们观察到出站通道在主题中开始发送2条消息，因为它应该只发送一条消息。一条带有实际json的消息将另一条消息的值计算为‘b’\x00\x06&‘字符。下面是代码和配置。如果删除transactionIdPrefix，那么我可以看到只发送给出
浏览 3提问于2021-10-01得票数 1
我有一个@StreamListener方法，它将执行REST调用。当REST调用返回异常时，@StreamListener方法抛出RunTimeException并重试。@StreamListener方法抛出RuntimeException时，重试次数不限
Spring云流重试配置：
spring.cloud.stream.kafka.bindings.inputChannel.consumer.enableDlq=true
spring.cloud.stream.bindings.inputChannel.consumer.maxAttempts=3
spring.cloud.stream.
浏览 0提问于2018-08-27得票数 1
我正在将一个使用Spring AMQP的项目迁移到一个使用Spring Cloud Stream和RabbitMQ的项目。 在我的旧项目中，当使用@RabbitListener处理消息时发生异常时，就会抛出该异常。如果绑定了死信队列，则仍会抛出异常(如果有重试，则仅抛出一次，我猜是最后一次)。这对于日志记录非常有帮助。 在Spring Cloud中，如果您定义以下属性，则@StreamListener有一个死信队列机制： spring.cloud.stream.bindings.input1.destination=dest1
spring.cloud.stream.rabbit.bindi
浏览 27提问于2019-01-19得票数 0
回答已采纳
这是应用程序。正如我在标题中提到的那样，StreamListener注释在StreamListener主类中起作用，但当它驻留在另一个带有@Component注释的类中时就不起作用了。
我认为这是某种StreamListener可见性问题。
我正在发布来自RabbitMQ管理接口的消息。
以下是我得到的例外：
ERROR 10676 --- [Service-Group-1] o.s.integration.handler.LoggingHandler : org.springframework.messaging.MessageDeliveryException: Dispatcher
浏览 2提问于2020-09-05得票数 1
回答已采纳
根据文档：https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/3.0.3.RELEASE/reference/html/spring-cloud-stream.html#_using_streamlistener_for_content_based_routing我可以根据如下条件将传入消息路由到处理程序： @EnableBinding(MySink.class)
@EnableAutoConfiguration
public static class TestPojoWithAnnotatedArguments {
浏览 70提问于2020-04-13得票数 0
我已经构建了一个应用程序，它通过kfka接收数据请求，并使用JPA来持久化数据，例如，在每个JPA错误上。
Exception in thread "user-4db12638-e58b-4728-9374-886ef20d0f31-StreamThread-1" org.springframework.dao.DataIntegrityViolationException: could not execute statement; SQL [n/a]; constraint [UK_sb8bbouer5wak8vyiiy4pf2bx]; nested exception is
浏览 0提问于2018-12-11得票数 0
我使用spring-cloud-stream-kafka-binder-3.0.4批量消费消息，消费后将消息转换为object，但得到上述异常。 代码如下： @StreamListener(ActivityChannel.ACTIVITY_INPUT_CHANNEL)
public void handleActivity(List<Message<Event>> messages,
@Header(name = "deliveryAttempt", defaultValue = "1") int deliveryAttempt,
浏览 64提问于2021-09-17得票数 0
回答已采纳
AFAIK是一个基于Spring Integration的Spring Cloud Stream项目。因此，我想知道是否有一种很好的方法来在StreamListener处理程序被触发之前重新排序入站消息的子集？或者，我是否需要使用XML或Spring Integration中的Java DSL config从头组装整个IntegrationFlow？
我的用例如下。大多数时候，我处理关于Kafka主题的入站消息。但是，必须根据CORRELATION_ID、SEQUENCE_NUMBER和SEQUENCE_SIZE标头对少数事件进行重排序。换句话说，我希望尽可能多地使用StreamListene
浏览 4提问于2018-11-22得票数 1
我有一个主题，在我的流处理器中有多个分区，我只想从一个分区中流式传输它，并且可以弄清楚如何配置它
spring.cloud.stream.kafka.streams.bindings.input.consumer.application-id=s-processor
spring.cloud.stream.bindings.input.destination=uinput
spring.cloud.stream.bindings.input.group=r-processor
spring.cloud.stream.bindings.input.contentType=application/
浏览 1提问于2019-02-16得票数 0
我有StreamListener，我想用新的功能模型和消费者<>替换它。不幸的是，我不知道如何将@Transactional转移到新模型： @Transactional
@StreamListener(PaymentChannels.PENDING_PAYMENTS_INPUT)
public void executePayments(PendingPaymentEvent event) throws Exception {
paymentsService.triggerInvoicePayment(event.getInvoiceId());
} 我已经累坏了某些东西
浏览 39提问于2021-10-24得票数 0
回答已采纳
我用春云条纹Kstream。我测试一个主题&一个@StreamListner。没事的。
我为两个KStream输入修改我的代码。(两个@StreamListener)但是，弹簧云错误..。
***************************
APPLICATION FAILED TO START
***************************
Description:
The bean 'stream-builder-process', defined in null, could not be registered. A bean with that
浏览 0提问于2019-03-13得票数 4
回答已采纳
我一直在尝试为Spring cloud stream kafka实现重试逻辑，以便在向主题sample-topic生成事件时抛出异常，它会再重试两次。
我在application.properties文件中添加了以下配置
spring.cloud.stream.bindings.processSampleEvent.destination=sample-topic
spring.cloud.stream.bindings.processSampleEvent.content-type=application/json
spring.cloud.stream.bindings.processSa
浏览 25提问于2021-07-06得票数 1
1回答
我需要一种根据传入消息的类型动态解析入站通道的方法。我不寻找任何基于头的解决方案，这已经在此链接https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/3.0.0.M1/spring-cloud-stream.html#_using_streamlistener_for_content_based_routing提到 必须根据消息的类型进行解析。如果有一个自定义绑定可以在应用程序启动时完成，就可以做到这一点；请给我一些如何实现这一点的示例。
浏览 13提问于2020-04-11得票数 0
在@StreamListener接收消息的方式(非功能性)中，会出现这样的情况：
@StreamListener(target = Sink.INPUT, condition = "headers['type']=='bogey'")
我现在要做的是：
spring.cloud.stream.function.definition=process spring.cloud.stream.bindings.process-in-0.destination=my-topic
@Bean
public Consumer<Message<So
浏览 4提问于2021-09-10得票数 0
我们有一个kafka过程，在这个过程中，我们使用来自一个主题的消息，然后会有一些充实发生，然后我们将消息发布到另一个主题。下面是一些事件
消费者-消费messageEnrichment -丰富了已消费的messageProducer -发布了其他主题的丰富信息。
我正在使用活页夹版本3.0.0-发行版，一切都很好。最近，我们引入了幂等生产者，并包括了transactionIdPrefix属性，我们发现我们开始出现性能问题。下面是统计数据。
在transactionIdPrefix系统之前花了大约20秒的时间来处理10k条消息，其中有30个并发用户。也就是说，每个second.After tra
浏览 5提问于2021-10-04得票数 0
我有个问题，消息来源发送了一个GenericMessage payload=xxxxx .接收器接收消息为10,120,120,120,120,120,120。
这个问题发生在我设置了Avro消息转换器之后。如果我删除Avro消息转换器并使用StreamListener来处理消息转换，它将很好地工作。
源application.properties
spring.cloud.stream.bindings.toGreeting.destination=greeting
spring.cloud.stream.bindings.toGreeting.contentType=application
浏览 13提问于2017-06-28得票数 0
回答已采纳
1回答
我的@StreamListener监听spring.cloud.stream.bindings.input.destination属性中指定的多个Kafka主题(完全类似于10个主题)。
有任何方法来确定来自@StreamListener注释方法内部的主题消息吗？我不想单独为所有主题配置绑定，编写10个几乎相同的StreamListeners。
浏览 0提问于2018-02-08得票数 2
回答已采纳
2回答
我有一个应用程序，我想要在卡夫卡消息中手动执行(n)ack。根据spring云文档，它应该使用autoCommitOffset 来完成。
但是，在我的应用程序中，即使定义了这样的属性，头KafkaHeaders.ACKNOWLEDGMENT仍然是空的。
下面是我的配置
spring.cloud.stream.kafka.binder.brokers=${KAFKA_BROKER_LIST}
spring.cloud.stream.default.contentType=application/json
spring.cloud.stream.bindings.mytopic.destinati
浏览 7提问于2020-05-14得票数 1
回答已采纳
我是卡夫卡的新手，我想要帮助
我有两个应用程序(springboot)，它们是相同的/拷贝只有不同的端口。
他们都是消费者
这两个人听XXX这个话题
我有另一个应用程序扮演制片人的角色。
每当我给某某话题发东西。
只有其中一个使用消息，而另一个不使用消息。
我分别测试了两种方法，如果他们是单独听的话，他们也会正常地听，但是如果他们在一起听，只有一人会听。
我在用
<dependency>
<groupId>org.springframework.cloud</groupId>
<a
浏览 0提问于2021-07-06得票数 1
回答已采纳
我们正在使用Spring Cloud Stream和Kakfa绑定，在我们的项目中，我们设想了下两个类来实现一个基础消费者： MyConsumer ==> StreamConsumer 其中StreamConsumer： public class StreamConsumer {
@StreamListener(EventSink.INPUT0)
public void handleMessages(Message<Notification> message) {
handle(message); // he
浏览 164提问于2019-06-27得票数 0
如何捕获写入Kafka主题的错误？
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.messaging.Message;
import org.springframework.messaging.handl
浏览 0提问于2016-11-06得票数 1
我想听两个Kafka主题，就像下面的代码一样，有两个源事件需要处理并转换为另一个事件。 所以我想在一个EnableBinding中监听这两个事件 @EnableBinding(PartnerOrderCancelledEventImporter.Targets.class)
public class PartnerOrderCancelledEventImporter {
@StreamListener(Targets.INPUT_ORDER_CANCELLED)
@SendTo(Targets.OUTPUT)
public K
浏览 50提问于2019-03-25得票数 0
回答已采纳
如何对反应性流管道进行错误处理。喜欢
应用程序错误处理(例如: errorChannel)
系统错误处理(使用DLQ、再处理等)
当前文档仅描述非反应性管道的错误处理。
Spring云流通过为错误处理场景提供简单的信任为用户做了相当简单的工作。如果同样的错误处理使用程序(具有相同的信任)工作在反应流管道上，那就太好了。用例和相应的配置细节如下：
@StreamListner("errorChannel")用于全局错误处理的注释
@KafkaListener(id="bar"，主题=“反应性-流-错误-主题”)
对于DLQ，并生
浏览 1提问于2018-11-22得票数 4
Spring版本1.3.3
我很好奇如何为一组绑定附加一个消息处理程序？对于每一个绑定，我都有完全相同的逻辑：
...
@StreamListener("topic_1")
void handleCouponCollectedMessage(FirstEventType event) {
eventsSet.add(event);
}
@StreamListener("topic_2")
void handleProfileUpdateMessage(SecondEventType event) {
浏览 0提问于2019-02-07得票数 0
1回答
你好，我正在写一个使用spring cloud stream的kafka消费者-生产者。在我的消费者内部，我将我的数据保存到数据库中，如果数据库关闭，我将手动退出应用程序.After重启应用程序，如果数据库仍然关闭，则应用程序再次停止。现在，如果我第三次重启应用程序，在中间时间间隔(两次失败)收到的消息丢失，kafka使用者将获取最新的消息，并跳过我退出代码的消息。 入站和出站通道绑定器接口 public interface EventChannel {
String inputEvent = "inputChannel";
String outputEvent =
浏览 33提问于2019-02-19得票数 1
1回答
新的消费者组从头开始挑选消息
、、、、
我有一个使用StreamListener注释的Kafka消费者，没有任何组id。现在，当我通过spring.cloud.stream.bindings.input.group添加group-id时，它从一开始就开始处理所有较旧的消息，这些消息在没有group-id时已经被这个消费者使用了。 如何将其配置为只挑选之前未挑选的未处理消息？ 我尝试将spring.cloud.stream.bindings.input.startOffset添加到latest，将resetOffsets添加到true，但都不起作用。 提前谢谢。:)
浏览 14提问于2021-01-21得票数 0
我正在使用Spring cloud streams消费rabbitmq的消息。我正在尝试获取一条失败的消息，使其在多次重试后仍保留在死信队列中。我在使用amqp之前已经通过编程做到了这一点，但是使用spring cloud streams似乎更难做到这一点。 @StreamListener(target = Sink.INPUT)
public void messageListener(final String in, @Header(name = "x-death", required = false) Map<?, ?> death) {
if (!va
浏览 17提问于2019-09-22得票数 0
回答已采纳
1回答
当我消耗了来自运动流的信息。我得到了一些带有头等的垃圾字符
@StreamListener(Processor.INPUT)
public void receive(String message) {
System.out.println("Message recieved: "+message);
throw new RuntimeException("Exception thrown");
}
@StreamListener("errorChannel")
浏览 1提问于2018-03-14得票数 2
回答已采纳
我能够使用代码发送和接收消息：
@EnableBinding(Processor.class)
public class KafkaStreamsConfiguration {
@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public String processMessage(String message) {
System.out.println("message = " + message);
return message.replaceAll("my",
浏览 1提问于2018-04-05得票数 1
回答已采纳
我有一个关于在处理批量(即batch-mode: true)时处理Spring Cloud Stream中的反序列化异常的问题。
根据这里的文档 (查看FailedFooProvider的实现)，该函数似乎应该返回原始消息的子类。
这里的意图是将Foo和BadFoo的列表都放在原始的@StreamListener方法中，然后由代码(即我)来对它们进行排序和单独处理？我怀疑是这样的，因为我已经读到了，因为它会重新提交整个批次。
如果是这样的话，如果应用程序通过不同的@StreamListener接收到一种以上的消息类型，比如Foo和Bar，那么在这种情况下value函数应该返回什么类型呢?下面是
浏览 46提问于2021-05-27得票数 0
回答已采纳
1回答
我使用和Spring。我的应用程序非常简单：
ExampleService.class：
@EnableBinding(Processor1.class)
@Service
public class ExampleService {
@StreamListener(Processor1.INPUT)
@SendTo(Processor1.OUTPUT)
public String dequeue(String message){
System.out.println("New message: " + message);
浏览 0提问于2019-01-09得票数 3
回答已采纳
@StreamListener(value = PersonStream.INPUT)
private void personBulkReceiver(List<Person> person) {
//....
}
spring:
cloud:
stream:
kafka:
binders:
bulkKafka:
type: kafka
environment:
spring:
cloud:
浏览 5提问于2020-11-16得票数 1
回答已采纳
2回答
我有以下简单的代码：
private int i = 0;
@StreamListener(Sink.INPUT)
public void processMessage(Message<?> message) {
i++;
}
@Scheduled(fixedDelay=5000)
private void scheduled(){
LOG.info("Messages consumed: " + i);
}
和下列属性：
spring.cloud.stream.bindi
浏览 4提问于2016-11-17得票数 0
回答已采纳
2回答
我正在使用Reactive Spring Cloud Stream，在创建没有输出的StreamListener时遇到了问题。只要没有收到格式错误的消息，下面的代码就可以工作。当接收到格式错误的消息时，通量关闭。 @StreamListener
public void handleMessage(@Input(MessagingConfig.INPUT) Flux<String> payloads) {
payloads.flatMap(objectToSave -> reactiveMongoTemplate.insert(objectToSave)).subsc
浏览 16提问于2019-01-04得票数 2
我需要使用spring cloud stream kafka绑定器实现一个幂等侦听器。
我们如何包装StreamListener，以便在将消息委托给相应的处理程序之前应用必要的过滤/重复数据删除建议？
谢谢。
浏览 4提问于2018-11-27得票数 1
1回答
我有两个用Java编写的微服务，使用Spring。，通过，在它们之间发送消息。我需要发送一个自定义头，但到目前为止没有成功。我在互联网和Spring 上已经阅读和尝试了大多数东西。
..。不过，我还是无法使它发挥作用。
这意味着我从未在接收方中接收到消息，因为无法找到标头，也不能为空。我怀疑标题从来没有写在消息中。现在我正试图用Kafkacat来验证这一点。
任何帮助都将是wellcome
提前谢谢。
--信息
这里是发件人代码：
@SendTo("notifications")
public void send(NotificationPayload payload, Str
浏览 5提问于2020-10-29得票数 0
1回答
我刚刚开始学习和Dataflow，我想知道一个重要的用例。我创建了示例处理器乘法器，它接收消息并将其重发5次输出。
@EnableBinding(Processor.class)
public class MultiplierProcessor {
@Autowired
private Source source;
private int repeats = 5;
@Transactional
@StreamListener(Processor.INPUT)
public void handle(String payload) {
浏览 0提问于2016-08-18得票数 2
回答已采纳
我在初始化StreamListener时遇到问题。我不能解决我的问题。 我在我的项目中使用了Spring Cloud Stream Kafka和Spring Cache。Spring Cache在SmartLifeCycle的start()方法之后初始化。但是StreamListener在SmartLifeCycle的start()方法之前就开始消耗数据了。因此，我无法在StreamListener开始使用数据之前完成缓存初始化，这给我带来了问题。 有什么方法可以延迟StreamListener处理或延迟StreamListener bean初始化吗？
浏览 23提问于2020-11-20得票数 1
回答已采纳
2回答
即使在特定配置之后，Spring流也不会在反序列化错误上重试。期望是，它应该基于配置的重试策略重试，最后将失败的消息推送到DLQ。
配置如下。
spring.cloud.stream.bindings.input_topic.consumer.maxAttempts=7
spring.cloud.stream.bindings.input_topic.consumer.backOffInitialInterval=500
spring.cloud.stream.bindings.input_topic.consumer.backOffMultiplier=10.0
spring.cloud.s
浏览 1提问于2019-06-26得票数 0
回答已采纳
1回答
如何执行重队列失败消息
、、、
根据docs ，目前支持的绑定器(、兔子和卡夫卡)依赖于RetryTemplate。以及GCP
我的项目细节
Spring 2.1.3版
依赖型pom.xml
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-gcp-starter-pubsub</artifactId>
<version>1.1.0.RELEASE</version>
</dependency>
浏览 0提问于2019-03-04得票数 2
所以我的问题是，我在我的属性文件中定义了一些Kafka主题，并且我可以在我的SpringBoot应用程序中从该主题读取KafkaStream<String, String>。但是我想访问KafkaStreams对象，这样我就可以打印我的KafkaStreams拓扑，这对开发很有用。在我的一个@StreamListener中，我尝试检索流生成器进程bean，这样我就可以以这种方式获取底层的KafkaStreams对象(如这里所描述的：https://cloud.spring.io/spring-cloud-stream-binder-kafka/spring-cloud-strea
浏览 46提问于2020-07-08得票数 0
我的任务是使用Spring Cloud Stream (使用Kafka绑定)。浏览Spring Cloud Stream的Quick Start，并从它使用的配置文件LogSinkConfiguration判断，这里的建议似乎是使用Spring Integration模式，例如@ServiceActivator(inputChannel = Sink.INPUT)来连接到输入通道。然而，我找到的所有教程都使用了来自Spring Cloud Stream库(vs Spring Integration)的一组不同的注释，即this演练中的@StreamListener(Processor.INP
浏览 24提问于2019-10-09得票数 1
回答已采纳
当Kinesis流使用者遇到错误，并在使用@StreamListener注释的方法中抛出异常时，如何配置绑定以使消息重新排队或发送到DLQ？我在Spring Cloud Stream中看到了对Rabbit和Kafka的这种级别的支持，但我没有看到对Kinesis绑定的支持。在Kinesis流处理器中处理错误的推荐方法是什么？它是如何通过配置和/或代码实现的？
浏览 24提问于2019-05-16得票数 1
回答已采纳
是否像描述的那样支持消息的请求
在我的@StreamListener中，我将抛出一个新的org.springframework.cloud.stream.binder.RequeueCurrentMessageException，但我在日志中看到org.springframework.integration.aws.inbound.kinesis.KclMessageDrivenChannelAdapter正在记录Consider to use 'errorChannel' flow for the compensation logic，然后在日志中看到的碎片被选中。在阅读了有
浏览 3提问于2020-11-18得票数 0
回答已采纳
1回答
我正在尝试设置一个简单的云流Sink，但始终会遇到以下错误。
我试过几个绑定器，它们都给出了相同的错误。
"SEVERE","logNameSource":"org.springframework.boot.diagnostics.LoggingFailureAnalysisReporter","message":"
***************************
APPLICATION FAILED TO START
***************************
Description:
Parame
浏览 0提问于2018-02-14得票数 3
我有一个使用spring cloud stream的消费者应用程序。这个应用程序正在使用队列中的消息(Kafka)，并且每条消息应用程序向4个不同的服务器发出4个不同的HTTP调用，其中一个非常慢(10秒应答)。当队列充满消息时，例如6000，应用程序会崩溃，原因有几个(1 - netty耗尽直接内存，2-我们正在使用reactor，线程池变空)。
有没有办法通过spring-cloud-stream或者kafka来限制消费端的消费速度？像每秒最大消息数这样的东西会更好。
这里可以看到kafka (application.yml)的配置。
spring:
kafka:
bootst
浏览 28提问于2020-02-12得票数 2
回答已采纳
我有一个关于spring-messaging (spring集成+ amqp)与RabbitMQ的问题。我有一个基于微服务架构的应用程序。一条服务发布消息，方式如下：
@EventListener(AfterCreationEvent.class)
@Publisher(channel = "messages")
public Message<Message> onCreateEntity(AfterCreationEvent<Message> event) {
if (isDisabled()) return null;
String
浏览 0提问于2017-04-09得票数 1
按照在GitHub https://github.com/spring-cloud/spring-cloud-gcp/tree/master/spring-cloud-gcp-samples/spring-cloud-gcp-pubsub-polling-binder-sample中找到的有关来自PubSub订阅的轮询消息的示例，我想知道... 是否可以让PollableMessageSource检索List<Message<?>>，而不是每次轮询一条消息？ 我见过只在Source类型的对象中使用@Poller表示法，而没有在Processor或Sink中使用。例如，
浏览 31提问于2021-01-14得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
腾讯会议活动推荐
腾讯云开发者
