作者回复: 首先，kafka 3副本在ack1时，极端情况里也是有可能会丢消息的，真不能丢的建议ack all。其次，我建议你还是默认认为消息是乱序的会重复的，虽然程序写起来要考虑的东西会多不少，但这样才够健壮。关于重复的问题，一般消息体里的内容总是会有什么业务ID的，让业务逻辑来做幂等性控制，这个不要强行放在MQ客户端想一劳永逸。

2019-05-23



10

york

事件驱动和消息驱动，是不是本质上是差不多的？应用程序内部，用Spring Event机制；而跨应用程序的场景，就用消息中间件？

作者回复: 如果我说消息也可以看成一种事件，你是不是就不会困惑了。我见过有自己给自己发个MQ消息的，也有很多人这么用。

2019-10-30



3

白菜炒五花肉

老师，直接用kafka-client和用spring cloud stream kafka有什么优劣，实际产线上哪个用的多啊

作者回复: 我感觉目前接触到的情况下，好像还是前者多一点，或者是spring-kafka，用spring cloud stream的少点

2021-08-14



2

jollyja

老师你好，咨询一个场景采用哪个消息队列产品及产品特性来实现比较合适：【场景描述】 ServiceSubscribeA订阅Message(ID=1)，线程阻塞（超时则抛异常）直到收到ID=1的Message（只收指定ID的消息）； ServiceSubscribeB订阅Message(ID=2)，线程阻塞（超时则抛异常）直到收到ID=2的Message（只收指定ID的消息）； ServicePublisher发布Message，ID=1、ID=2、。。。 subscribe和publish，无先后关系； subscribe并发数要求高，可能同时几千个线程（不同jvm实例）做subscribe； subscribe处理时效要求高，不考虑publish耗时的话，从subscribe开始到收到消息的耗时为毫秒级；

作者回复: 订阅消息为什么会有线程阻塞，没收到消息的时候监听队列就可以了呀，不用做什么阻塞。看你的描述两个消费者订阅的是同一个队列，只是各收不同的消息，JMS规范里有message selector，可以根据消息头里的不同属性来筛选消息。但我总觉得你可能没把需求说清楚，如果是不同的系统，通常监听不同的队列，如果监听同一个队列，那也是不同的分组，消息都会收下来的。感觉没能理解你的需求。

2021-05-04



2

test

老师，如果kafka要自己手动提交offset是不是就不能按照你的例子这样子写。

作者回复: 那可以直接使用Spring Kafka，不用Spring Cloud Stream的封装

2020-08-07



2

高峰

你好，这种binder的模式，如何设置 consumer group-Id

作者回复: 我们在kafka-barista-service的配置文件中有一个spring.cloud.stream.bindings.newOrders.group=barista-service，这里的newOrders就是Topic，group是分组名

2020-07-23

2

2

十块钱

kafka之前做过日志分析，现在公司用阿里rocketmq，使用起来比较方便，但是要做好消息幂等性

2019-05-21



2

高峰

请教一下老师，我这边单点kafka,单个消费者，单个partion, 我用spring boot kafka binder 消费一段时间后自动退出了，报一下错误。 Revoke previously assigned partitions xxxxxx Member consumer-xxxx-consumer-01-2-e7d8f7bf-ea52-4b3a-8aed-39059da4f772 sending LeaveGroup request to coordinator 172.16.12.170:9092 (id: 2147483647 rack: null) due to the consumer unsubscribed from all topics 是不是长时间没有数据可以消费导致的。 allow.auto.create.topics = true auto.commit.interval.ms = 100 auto.offset.reset = earliest bootstrap.servers = [172.16.12.170:9092] check.crcs = true client.dns.lookup = default client.id = client.rack = connections.max.idle.ms = 540000 default.api.timeout.ms = 60000 enable.auto.commit = false exclude.internal.topics = true fetch.max.bytes = 52428800 fetch.max.wait.ms = 500 fetch.min.bytes = 1 group.id = hot-consumer-01 group.instance.id = null heartbeat.interval.ms = 3000 interceptor.classes = [] internal.leave.group.on.close = true isolation.level = read_uncommitted key.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer max.partition.fetch.bytes = 1048576 max.poll.interval.ms = 1200000 max.poll.records = 500 metadata.max.age.ms = 300000 metric.reporters = [] metrics.num.samples = 2 metrics.recording.level = INFO metrics.sample.window.ms = 30000 partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor] receive.buffer.bytes = 65536 reconnect.backoff.max.ms = 1000 reconnect.backoff.ms = 50 request.timeout.ms = 30000 retry.backoff.ms = 100

2020-09-23





收起评论