如果有多条消息，如何测试Spring Cloud Stream？
Spring Cloud Stream是一个用于构建消息驱动微服务的框架。它基于Spring Boot，提供了一种简单且灵活的方式来连接消息代理和应用程序。当我们需要测试Spring Cloud Stream时，可以按照以下步骤进行：
- 创建测试类：首先，我们需要创建一个测试类来编写测试代码。可以使用JUnit或其他测试框架来编写测试用例。
- 模拟消息代理：为了测试Spring Cloud Stream，我们需要模拟消息代理。可以使用Mockito或其他模拟框架来模拟消息代理的行为。
- 定义输入和输出通道：在测试类中，我们需要定义输入和输出通道。输入通道用于发送消息，输出通道用于接收消息。
- 发送消息：使用输入通道发送消息。可以使用Spring Cloud Stream提供的注解或API来发送消息。
- 验证消息：使用输出通道接收消息，并验证消息的内容和格式是否符合预期。可以使用断言来验证消息。
- 执行测试：运行测试类，执行测试代码。可以使用Maven或其他构建工具来运行测试。
- 分析测试结果：分析测试结果，检查是否有错误或异常。如果有错误或异常，可以根据错误信息进行调试和修复。
Spring Cloud Stream的测试可以帮助我们验证消息的传递和处理是否正确，以及确保应用程序在消息驱动的环境中正常工作。
对于Spring Cloud Stream的更多信息和使用示例，可以参考腾讯云的相关产品和文档：
- 腾讯云消息队列 CMQ：提供了可靠的消息传递服务，适用于构建分布式系统和微服务架构。详情请参考：腾讯云消息队列 CMQ
- 腾讯云云原生应用引擎 TKE：提供了一种简单且高效的方式来部署和管理容器化应用程序。详情请参考：腾讯云云原生应用引擎 TKE
请注意，以上链接仅为示例，实际使用时应根据具体需求选择适合的产品和服务。
相关·内容
我们进行了集成测试，最终将多条消息发送到同一个目的地，但有许多不同的组。在rabbitMQ术语中，相同的交换具有不同的主题。
spring.cloud.stream.bindings.userEdited-out-0.destination=user.exchange
spring.cloud.stream.rabbit.bindings.userEdited-out-0.producer.routing-key-expression='user-edited.v1'
spring.cloud.stream.bindings.userEdited-out-0.content-t
浏览 4提问于2022-05-12得票数 1
回答已采纳
我有一个基于微服务的应用程序，从卡夫卡主题读取消息。当服务关闭时，如果有任何关于该主题的消息正在写入，我希望使用者在下次启动并运行时读取这些消息。但是当服务中断时，我错过了所有的消息。如何让使用者读取在服务关闭时未被读取的消息？
当我的微服务启动时，我收到了所有的消息，所有的消息都返回到了主题。
我的application.properties：
spring.cloud.stream.bindings.input.destination=test
spring.cloud.stream.bindings.input.consumer.headerMode=raw
spring.cloud.s
浏览 1提问于2017-07-29得票数 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
我正在尝试Spring Cloud Stream，当我试图运行最简单的测试之一时(原始测试可以在这里找到：https://github.com/spring-cloud/spring-cloud-stream/blob/master/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/disable/AutoconfigurationDisabledTest.java) 使用:Java1.8Spring 2.1.5.RELEASE SpringBoot -cloud.vers
浏览 10提问于2019-05-22得票数 0
1回答
我有两个用Java编写的微服务，使用Spring。，通过，在它们之间发送消息。我需要发送一个自定义头，但到目前为止没有成功。我在互联网和Spring 上已经阅读和尝试了大多数东西。
..。不过，我还是无法使它发挥作用。
这意味着我从未在接收方中接收到消息，因为无法找到标头，也不能为空。我怀疑标题从来没有写在消息中。现在我正试图用Kafkacat来验证这一点。
任何帮助都将是wellcome
提前谢谢。
--信息
这里是发件人代码：
@SendTo("notifications")
public void send(NotificationPayload payload, Str
浏览 5提问于2020-10-29得票数 0
我能够使用代码发送和接收消息：
@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
回答已采纳
我创建了一个示例HttpSource和HttpSink。我的application.properties看起来像这样
Source
spring.cloud.stream.bindings.output.destination=greetings
spring.cloud.stream.bindings.output.partitionKeyExpression=payload
spring.cloud.stream.bindings.output.partitionCount=2
Sink
spring.cloud.stream.bindings.input.destination=gr
浏览 4提问于2016-08-16得票数 0
回答已采纳
1回答
我正在使用与kafka经纪公司的春云流进行微服务互传。作为其中的一部分，stream bridge将用于发送消息，这是很好的。
--但是在使用所述消息期间，不需要立即使用该消息，而是当满足条件时，只需要使用。
从文档中，我了解到我需要在这方面使用Polled Consumers(如果我弄错了就纠正我)。
这就是我对的理解所做的尝试。
application.properties
spring.cloud.stream.pollable-source = consumeResponse
spring.cloud.stream.function.definition = consumeRespon
浏览 7提问于2022-03-10得票数 0
回答已采纳
我在RabbitMQ中使用spring云流。(Spring版本是格林威治)我有一个有两个队列的主题，我想使用路由密钥向每个队列发送消息。
但它不能正常工作。
当我通过每个@输出通道发送消息时，消息会同时发送到两个队列。(我想要一个队列的1条消息，但1条消息转到2个队列)
我的生产者配置如下(实际代码在yaml中)
Spring.cloud.stream.bindings.output1.destination=topic-a
Spring.cloud.stream.bindings.output1.producer.requiredGroups=queue1
Spring.cloud.stre
浏览 0提问于2019-06-10得票数 0
回答已采纳
4回答
下面是基于应用程序的测试场景。我的应用程序有一个有两个队列的主题。第一个队列的BindingKey被命名为“城市”，因为第二个队列是绑定键"persons“。
请如何为春云流兔生产者设置路由密钥？去调查信息将被消耗在哪里？
这是我的绑定配置：
spring.config.name=streaming
spring.cloud.stream.bindings.citiesChannel.destination=streamInput
spring.cloud.stream.bindings.citiesChannel.group=cities
spring.cloud.stream.r
浏览 0提问于2017-04-04得票数 3
回答已采纳
1回答
我们有一个需求，当我们使用来自一个主题的消息时，会有一些充实发生，然后我们将消息发布到另一个主题。下面是一些事件
消费者-消费messageEnrichment -丰富了已消费的messageProducer -发布了其他主题的丰富信息。
我正在使用活页夹，一切都很好。突然，我们观察到生产者正在向主题发送重复的信息，然后我们使生产者是幂等的。为了更好的控制，我们让autocommitOffSet做错了。下面是我们在这个方法中所做的工作
@StreamListener("INPUT")
@SendTo("OUTPUT")
public void consumer(
浏览 3提问于2021-09-28得票数 0
1回答
我正在使用和注册Avro模式。
架构已成功注册。但是当我的Stream侦听器接收到一条消息时，有效负载仍然在Bytes中。
这是我的财产。
spring.cloud.stream.schemaRegistryClient.endpoint= spring.cloud.stream.bindings.output.contentType=application/*+avro spring.cloud.stream.schema.avro.dynamic-schema-generation-enabled=true spring.cloud.stream.schemaRegistryClient
浏览 4提问于2017-08-22得票数 2
2回答
我想知道在交换中使用未知/无效路由键值处理消息的最佳方法是什么？在我的例子中，我在同一个交换中发送我的所有消息，并且基于路由密钥，消息被路由到相应的队列。下面是我的配置(我正在使用Stream)：
spring.cloud.stream.bindings.output.destination: my-exchange
spring.cloud.stream.bindings.output.producer.routingKeyExpression: payload.type
spring.cloud.stream.bindings.input-type1.destination: my-ex
浏览 1提问于2020-12-03得票数 0
回答已采纳
我们使用spring-cloud-stream来管理应用程序之间的消息。
我们有自定义绑定：
public interface InboundChannels {
String TASKS = "domainTasksInboundChannel";
String EVENTS = "eventsInboundChannel";
@Input(TASKS)
SubscribableChannel tasks();
@Input(EVENTS)
SubscribableChannel events();
}
public inte
浏览 0提问于2018-04-07得票数 2
我正在尝试在spring cloud stream中实现延迟消息(不使用rabbitmq插件)，但它不起作用
我使用spring-boot实现了它，它工作得很好。下面是我在spring-boot中做的示例代码。
我正在尝试在spring-cloud-stream中做同样的事情，但是没有帮助。下面是属性。
输出通道-生产者
spring.cloud.stream.bindings.output.destination=temp-channel
spring.cloud.stream.bindings.output.group=temp-channel-group
输入渠道-消费者
spring
浏览 18提问于2019-08-14得票数 0
1回答
当使用动态生成的目标时，我希望禁用消息中的嵌入标头，类似于来自的示例(例如输出主题= dyntopic1、dyntopic2、.)。
我已经设置了像下面这样的属性，但是我仍然得到标题，如果我遗漏了什么，有什么建议吗？
spring.cloud.stream.bindings.output.group=test-ogroup
spring.cloud.stream.bindings.output.binder=kafka
spring.cloud.stream.bindings.output.producer.headerMode=raw
spring.cloud.stream.bindings
浏览 4提问于2018-01-31得票数 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
回答已采纳
1回答
我有多个生产者将消息推送到我的kafka代理中。我想保护我的主题，这样生产者需要密码才能在主题中推送消息或任何形式的保护。
在spring boot中，可以看到以下属性。
spring.cloud.stream.kafka.binder.configuration.[security.protocol]=SSL
#spring.cloud.stream.kafka.binder.configuration.[ssl.truststore.location]=D:\\streaming\\MessagingQueingTier\\SSL\\kafka.client.truststore.jks
浏览 0提问于2020-08-22得票数 2
我正在构建一个基于SpringCloud流的应用程序，交换类型是主题，消息从主题交换发送到2个队列消费者组。场景是这样的：
我的应用程序中的服务A希望根据不同的用例场景(如预订、取消、更新等)，通过名为: appointments request的交换向服务B和服务C发送约会类型的消息。
因此，带有密钥appointments.book.B或Appointments.ancel.B的消息应转到消费者队列组appointments.B
带有密钥appointments.book.C或appointments.ancel.C的消息应转到消费者队列组约会。C
如何成功地实现这一点？
生产者服务的配置
浏览 11提问于2019-07-10得票数 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
回答已采纳
我正在编写一个生产者来使用Spring cloud stream库将消息推送到Kinesis stream。我能够成功地将数据推送到kinesis，但在kinesis方面，它失败了，吞吐量超过了exception.Is。有什么方法可以重试再次推送这些消息，并且确切地知道哪条消息失败了？另外，我不想使用KPL或KCL。
我尝试了答案中建议的解决方案，以下是我的配置：
spring.cloud.stream.bindings.input.producer.errorChannelEnabled: true spring.cloud.stream.bindings.input.producer.er
浏览 5提问于2018-12-02得票数 1
2回答
我正在提供新的软件设计，并将使用RabbitMQ作为消息总线以及Spring Cloud Stream。
其中一个问题是开发人员无法在本地计算机上安装Erlang或RabbitMQ。
Spring Cloud stream不支持ActiveMQ。在unix环境中使用RabbitMQ时，有没有什么方法可以在本地机器上安装队列并使用它？
浏览 6提问于2018-10-04得票数 1
1回答
处理死信队列的配置
、、、、
我有一个使用RabbitMQ在微服务中交换消息的项目。对我的项目至关重要的一件事是，我绝不能失去任何信息。
为了尽量减少失败，我计划如下：
对队列中的消息使用默认的重试方法
配置死信队列，以便在一段时间后再将消息放到队列中。
为了避免无限循环，只允许几次(比方说，5)消息可以从死信队列重新发布到常规消息队列。
我相信我可以使用下面的配置实现前两项：
#dlx/dlq setup - retry dead letter 5 minutes later (300000ms later)
spring.cloud.stream.rabbit.bindings.input.cons
浏览 3提问于2019-12-09得票数 0
回答已采纳
我有一个使用Spring Cloud Stream接收消息的消费者应用程序。我已配置重试和DLQ，但在超过重试次数后获取MessagingException。
下面是我的配置：
spring.cloud.stream.bindings.input.group = myProcessingGroup
spring.cloud.stream.bindings.input.consumer.max-attempts=3
spring.cloud.stream.rabbit.bindings.input.consumer.auto-bind-dlq=true
spring.cloud.stream.r
浏览 0提问于2019-08-26得票数 0
如何消费来自pubsub的多条消息？这似乎是一个简单的问题，应该有简单的解决方案，但目前我可以找到使用spring-cloud-gcp-pubsub从pubsub中使用批量记录的简单方法。
我使用spring-cloud-gcp-pubsub来消费来自pubsub的消息，并在spring boot应用程序中处理它们。我目前的设置非常简单，我有使用记录的PubSubInboundChannelAdapter和ServiceActivator。经过研究，我发现了spring集成Aggregators，但它们似乎不是一个很好的方法，因为它不容易向下游传播确认。我是不是漏掉了什么？如何批量消费消息？
浏览 0提问于2019-08-30得票数 0
1回答
任何想法都将不胜感激，我正在试着写测试；
@ExtendWith(SpringExtension.class)
@EmbeddedKafka(count = 1, controlledShutdown = true, topics = { "input", "output" }, brokerProperties = { "broker.id=2",
"listeners=PLAINTEXT://127.0.0.1:9092" })
class BindingTest {
@Autowired
private Appli
浏览 1提问于2018-11-15得票数 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
回答已采纳
使用主题交换配置多个队列，并使用路由键使用spring云流引导消息特定队列
我的要求是示例，我在使用者端将队列和exchange定义为下面的内容
spring.cloud.stream.bindings.inputA.destination=Common-Exchange
spring.cloud.stream.bindings.inputA.group=A-Queue
spring.cloud.stream.bindings.inputB.destination=Common-Exchange
spring.cloud.stream.bindings.inputB.group=B-Queu
浏览 0提问于2019-05-11得票数 0
回答已采纳
2回答
在特定主题分区中发送消息
、、、
生产者有没有办法将消息发送到代理中特定的主题部分？ 到目前为止，我可以发送有2个分区的主题，但不能控制发送到特定的分区。 Component
@EnableBinding(Source.class)
public class RsvpsKafkaProducer {
private static final int SENDING_MESSAGE_TIMEOUT_MS = 10000;
private final Source source;
public RsvpsKafkaProducer(Source source) {
this.so
浏览 25提问于2020-10-22得票数 0
回答已采纳
我正在尝试用不同的主题发送消息，然后在它想要听的主题上配置使用者。
我的想法是使用单一目标"domainMessage“并使用自定义分区策略。我有一个枚举，我只是使用这个值，因为partitionKey和partitionStrategy只返回键(假设键总是等于生产者端的分区计数)。
这个能行吗？如果是这样的话，我不知道如何配置使用者。
我的制作人有下面的application.properties
spring.cloud.stream.bindings.output.destination=domainMessages
spring.cloud.stream.bindings.ou
浏览 1提问于2016-09-13得票数 0
回答已采纳
我遇到了这样的情况，我想暂停消费者停止通过spring boot应用程序消费消息，并且消息不应该从队列中丢失，我正在使用amqp和spring cloud stream。 https://github.com/spring-cloud/spring-cloud-stream/blob/master/docs/src/main/asciidoc/spring-cloud-stream.adoc#binding_visualization_control声明只有kafka有这个选项。 我能为rabbit mq活页夹做些什么吗？
浏览 44提问于2019-03-31得票数 0
我知道使用Kafka Connect，您可以创建连接器，将所有kafka消息汇聚到一个PUBSUB主题。
我想知道我是否可以将Spring Cloud Stream绑定器与Spring Cloud Function结合使用，并将所有这些部署到Google Functions。
根据我的理解，我可以将Spring Cloud Stream和Spring Cloud Function结合起来。这是否意味着我可以使用Spring Cloud Stream中的绑定器来实际接受来自Kafka的消息？
浏览 1提问于2020-09-08得票数 0
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
我正在使用avro生成一个java类( Heartbeat )，并且我正在使用Spring云消息处理器，以便通过此Heartbeat类将消息推送到kafka。
所以这是我的服务：
@Service
public class HeartbeatServiceImpl implements HeartbeatService {
private Processor processor;
public HeartbeatServiceImpl(Processor processor) {
this.processor = processor;
}
@Override
浏览 1提问于2021-05-28得票数 0
我在应用程序A中有producer，配置如下，
生产者属性：
spring.cloud.stream.bindings.packageVersionUpdatesPublishChannel.destination=fabric-exchange
spring.cloud.stream.bindings.packageVersionUpdatesPublishChannel.producer.requiredGroups=version-updates
spring.cloud.stream.rabbit.bindings.packageVersionUpdatesPublis
浏览 1提问于2019-05-30得票数 0
我有一个基于Spring的绑定应用程序。它定义了一个包含以下部分的拓扑：
绿色中的数字显示了通过通过流绑定绑定的各个处理器定义的拓扑传递的消息数量，以下是各自的属性：
spring.cloud.stream.bindings:
...
hint1Stream-out-0:
destination: hints
realityStream-out-0:
destination: hints
countStream-in-0:
destination: hints
我正在计算每个处理器使用peek()方法生成/使用的消息如下：
return str
浏览 1提问于2020-05-30得票数 2
回答已采纳
我们有一个kafka过程，在这个过程中，我们使用来自一个主题的消息，然后会有一些充实发生，然后我们将消息发布到另一个主题。下面是一些事件
消费者-消费messageEnrichment -丰富了已消费的messageProducer -发布了其他主题的丰富信息。
我正在使用活页夹版本3.0.0-发行版，一切都很好。最近，我们引入了幂等生产者，并包括了transactionIdPrefix属性，我们发现我们开始出现性能问题。下面是统计数据。
在transactionIdPrefix系统之前花了大约20秒的时间来处理10k条消息，其中有30个并发用户。也就是说，每个second.After tra
浏览 5提问于2021-10-04得票数 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回答
在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
我使用的是Spring Boot2.1.1.RELEASE和Spring Cloud Greenwich.RC2，spring-cloud-stream-binder-kafka的托管版本是2.1.0RC4。Kafka版本为1.1.0。我设置了以下属性，因为如果出现错误，消息将不会被使用。 spring.cloud.stream.bindings.input.group=consumer-gp-1
...
spring.cloud.stream.kafka.bindings.input.consumer.autoCommitOnError=false
spring.cloud.stream.
浏览 29提问于2019-03-18得票数 2
尝试使用绑定库从gcp发布子主题轮询消息。的参考文献。一次使用spring.cloud.stream.gcp.pubsub.default.consumer.maxFetchSize轮询获取N条消息( maxFetchSize的值)，但甚至我也设置了该属性，并将maxFetchSize设置为2或其他。我已经收到了一条来自主题的消息，甚至主题还有其他消息。这里有人有别的主意吗？
春季云版本: 2021.0.3
谷歌spring-云-gcp-依赖版本: 3.3.0
java: openjdk 11
# application.properties
spring.cloud.gc
浏览 11提问于2022-07-19得票数 0
我有一个rabbitmq exchange test-exchange，它是使用路由关键字test- routing -key绑定到队列test-queue的直接交换。
我正尝试使用spring云流将消息发布到此交换，但消息未到达队列
我在pom.xml中的依赖
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</depende
浏览 13提问于2019-08-16得票数 1
版本: Spring Boot: 1.4.2发布Spring Cloud Deps: Brixton.SR7
这是我的一个处理器应用的application.properties。
logging.level.=DEBUG
server.port=0
logging.file=traveller-events-processor.log
server.port=0
spring.cloud.stream.rabbit.bindings.input.consumer.bindingRoutingKey='aa'
spring.cloud.stream.rabbit.binding
浏览 3提问于2017-07-20得票数 1
回答已采纳
我正在尝试创建一个Spring契约，以便在使用时测试消息。我已经在pom.xml中包含了依赖项
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>
<!--TEST-->
<dependency>
<groupId>org.spring
浏览 0提问于2020-03-09得票数 0
回答已采纳
我正在尝试使用spring云流供应商和消费者来实现上述场景。
这个应用程序是一个包含生产者的春季引导应用程序，consumer.There是一个生产者，可以是多个消费者。所有消费者都应该表现为一个客户端来排队(即单条消息只由单个消费者接收)，而其他消费者则接收不同的消息。
下面是java类
@Component
public class MultipleFunctionsApplication {
@Bean
public Consumer<String> sink1() {
return message -> {
浏览 4提问于2021-08-27得票数 0
回答已采纳
我是卡夫卡的新手，我想要帮助
我有两个应用程序(springboot)，它们是相同的/拷贝只有不同的端口。
他们都是消费者
这两个人听XXX这个话题
我有另一个应用程序扮演制片人的角色。
每当我给某某话题发东西。
只有其中一个使用消息，而另一个不使用消息。
我分别测试了两种方法，如果他们是单独听的话，他们也会正常地听，但是如果他们在一起听，只有一人会听。
我在用
<dependency>
<groupId>org.springframework.cloud</groupId>
<a
浏览 0提问于2021-07-06得票数 1
回答已采纳
2回答
我按照前面的问题和配置的流进行了描述，但是，我无法使它正确工作。
我的设置如下。我有两个应用程序A和B。App A使用输入通道one，输出two。App B使用输入two。通道two配置为内容类型application/json。
应用程序A.属性。
spring.cloud.stream.bindings.input.destination=one
spring.cloud.stream.bindings.input.group=default
spring.cloud.stream.bindings.output.destination=two
spring.cloud.stream.b
浏览 0提问于2016-03-15得票数 3
回答已采纳
我想每15分钟批量消费一次消息。为此，我设置了这些属性， spring.cloud.stream.kafka.binder.consumer-properties.max.poll.records=5000000
spring.cloud.stream.kafka.binder.consumer-properties.fetch.max.wait.ms=900000
spring.cloud.stream.kafka.binder.consumer-properties.fetch.min.bytes=500000000 当我将此属性spring.cloud.stream.kafka.bin
浏览 131提问于2021-09-22得票数 0
回答已采纳
我有一个可以侦听单个RabbitMQ队列的工作应用程序。
但是，当我添加另一个消耗消息的bean并试图将其绑定到另一个队列时，这两个队列都不是在RabbitMQ中创建的，手动创建时也不会从这些队列中使用任何消息。
我创建了一个小kotlin项目来演示这个问题：
@SpringBootApplication
class SpringCloudStreamTwoRabbitConsumersApplication
fun main(args: Array<String>) {
runApplication<SpringCloudStreamTwoRabbitConsum
浏览 1提问于2021-03-09得票数 0
回答已采纳
我将Spring Cloud Stream与Avro和Confluent Schema Registry结合使用。我对所有服务都使用一个DLQ主题，因此具有不同模式的消息可能会落在此主题中。我禁用了动态模式注册，以确保不会通过不正确的消息(schemaspring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled= false)。
然而，这个问题是由于dlq上缺乏模式造成的，我可能会因为登录到这个主题而丢失一条消息。因此，我希望能够将JSON格式的消息生成到dlq，并将Avro用于管道的其余部分。如果有人能帮助我如何实现这一点，或者
浏览 8提问于2019-08-09得票数 1
云服务器
ICP备案
对象存储
腾讯会议
实时音视频
腾讯云开发者
