diff --git a/labx-10/labx-10-sc-stream-rabbitmq-consumer-ack/pom.xml b/labx-10/labx-10-sc-stream-rabbitmq-consumer-ack/pom.xml new file mode 100644 index 00000000..2e3c1fe7 --- /dev/null +++ b/labx-10/labx-10-sc-stream-rabbitmq-consumer-ack/pom.xml @@ -0,0 +1,58 @@ + + + + labx-10 + cn.iocoder.springboot.labs + 1.0-SNAPSHOT + + 4.0.0 + + labx-10-sc-stream-rabbitmq-consumer-ack + + + 1.8 + 1.8 + 2.2.4.RELEASE + Hoxton.SR1 + + + + + + + org.springframework.boot + spring-boot-starter-parent + ${spring.boot.version} + pom + import + + + org.springframework.cloud + spring-cloud-dependencies + ${spring.cloud.version} + pom + import + + + + + + + + org.springframework.boot + spring-boot-starter-web + + + + + org.springframework.cloud + spring-cloud-starter-stream-rabbit + + + + diff --git a/labx-10/labx-10-sc-stream-rabbitmq-consumer-ack/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/consumerdemo/ConsumerApplication.java b/labx-10/labx-10-sc-stream-rabbitmq-consumer-ack/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/consumerdemo/ConsumerApplication.java new file mode 100644 index 00000000..85330de4 --- /dev/null +++ b/labx-10/labx-10-sc-stream-rabbitmq-consumer-ack/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/consumerdemo/ConsumerApplication.java @@ -0,0 +1,16 @@ +package cn.iocoder.springcloud.labx10.rabbitmqdemo.consumerdemo; + +import cn.iocoder.springcloud.labx10.rabbitmqdemo.consumerdemo.listener.MySink; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.cloud.stream.annotation.EnableBinding; + +@SpringBootApplication +@EnableBinding(MySink.class) +public class ConsumerApplication { + + public static void main(String[] args) { + SpringApplication.run(ConsumerApplication.class, args); + } + +} diff --git a/labx-10/labx-10-sc-stream-rabbitmq-consumer-ack/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/consumerdemo/listener/Demo01Consumer.java b/labx-10/labx-10-sc-stream-rabbitmq-consumer-ack/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/consumerdemo/listener/Demo01Consumer.java new file mode 100644 index 00000000..2009a2f8 --- /dev/null +++ b/labx-10/labx-10-sc-stream-rabbitmq-consumer-ack/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/consumerdemo/listener/Demo01Consumer.java @@ -0,0 +1,39 @@ +package cn.iocoder.springcloud.labx10.rabbitmqdemo.consumerdemo.listener; + +import cn.iocoder.springcloud.labx10.rabbitmqdemo.consumerdemo.message.Demo01Message; +import com.rabbitmq.client.Channel; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.amqp.support.AmqpHeaders; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.messaging.handler.annotation.Header; +import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.stereotype.Component; + +import java.io.IOException; +import java.util.concurrent.atomic.AtomicInteger; + +@Component +public class Demo01Consumer { + + private Logger logger = LoggerFactory.getLogger(getClass()); + + private AtomicInteger index = new AtomicInteger(); + + @StreamListener(MySink.DEMO01_INPUT) + public void onMessage(@Payload Demo01Message message, + @Header(AmqpHeaders.CHANNEL) Channel channel, + @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException { + logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message); + // 提交消费进度 +// if (message.getId() % 2 == 1) { + if (index.incrementAndGet() % 2 == 1) { + // ack 确认消息 + // 第二个参数 multiple ，用于批量确认消息，为了减少网络流量，手动确认可以被批处。 + // 1. 当 multiple 为 true 时，则可以一次性确认 deliveryTag 小于等于传入值的所有消息 + // 2. 当 multiple 为 false 时，则只确认当前 deliveryTag 对应的消息 + channel.basicAck(deliveryTag, false); + } + } + +} diff --git a/labx-10/labx-10-sc-stream-rabbitmq-consumer-ack/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/consumerdemo/listener/MySink.java b/labx-10/labx-10-sc-stream-rabbitmq-consumer-ack/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/consumerdemo/listener/MySink.java new file mode 100644 index 00000000..7a894faa --- /dev/null +++ b/labx-10/labx-10-sc-stream-rabbitmq-consumer-ack/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/consumerdemo/listener/MySink.java @@ -0,0 +1,13 @@ +package cn.iocoder.springcloud.labx10.rabbitmqdemo.consumerdemo.listener; + +import org.springframework.cloud.stream.annotation.Input; +import org.springframework.messaging.SubscribableChannel; + +public interface MySink { + + String DEMO01_INPUT = "demo01-input"; + + @Input(DEMO01_INPUT) + SubscribableChannel demo01Input(); + +} diff --git a/labx-10/labx-10-sc-stream-rabbitmq-consumer-ack/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/consumerdemo/message/Demo01Message.java b/labx-10/labx-10-sc-stream-rabbitmq-consumer-ack/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/consumerdemo/message/Demo01Message.java new file mode 100644 index 00000000..74c4255d --- /dev/null +++ b/labx-10/labx-10-sc-stream-rabbitmq-consumer-ack/src/main/java/cn/iocoder/springcloud/labx10/rabbitmqdemo/consumerdemo/message/Demo01Message.java @@ -0,0 +1,29 @@ +package cn.iocoder.springcloud.labx10.rabbitmqdemo.consumerdemo.message; + +/** + * 示例 01 的 Message 消息 + */ +public class Demo01Message { + + /** + * 编号 + */ + private Integer id; + + public Demo01Message setId(Integer id) { + this.id = id; + return this; + } + + public Integer getId() { + return id; + } + + @Override + public String toString() { + return "Demo01Message{" + + "id=" + id + + '}'; + } + +} diff --git a/labx-10/labx-10-sc-stream-rabbitmq-consumer-ack/src/main/resources/application.yml b/labx-10/labx-10-sc-stream-rabbitmq-consumer-ack/src/main/resources/application.yml new file mode 100644 index 00000000..92bfb742 --- /dev/null +++ b/labx-10/labx-10-sc-stream-rabbitmq-consumer-ack/src/main/resources/application.yml @@ -0,0 +1,35 @@ +spring: + application: + name: demo-consumer-application + cloud: + # Spring Cloud Stream 配置项，对应 BindingServiceProperties 类 + stream: + # Binder 配置项，对应 BinderProperties Map + binders: + rabbit001: + type: rabbit # 设置 Binder 的类型 + environment: # 设置 Binder 的环境配置 + # 如果是 RabbitMQ 类型的时候，则对应的是 RabbitProperties 类 + spring: + rabbitmq: + host: 127.0.0.1 # RabbitMQ 服务的地址 + port: 5672 # RabbitMQ 服务的端口 + username: guest # RabbitMQ 服务的账号 + password: guest # RabbitMQ 服务的密码 + # Binding 配置项，对应 BindingProperties Map + bindings: + demo01-input: + destination: DEMO-TOPIC-01 # 目的地。这里使用 RabbitMQ Exchange + content-type: application/json # 内容格式。这里使用 JSON + group: demo01-consumer-group-DEMO-TOPIC-01 # 消费者分组 + binder: rabbit001 # 设置使用的 Binder 名字 + # RabbitMQ 自定义 Binding 配置项，对应 RabbitBindingProperties Map + rabbit: + bindings: + demo01-input: + # RabbitMQ Consumer 配置项，对应 RabbitConsumerProperties 类 + consumer: + acknowledge-mode: MANUAL # 消费消息的确认模式，默认为 AUTO 自动确认 + +server: + port: ${random.int[10000,19999]} # 随机端口，方便启动多个消费者 diff --git a/labx-10/pom.xml b/labx-10/pom.xml index d81a3be1..100dc550 100644 --- a/labx-10/pom.xml +++ b/labx-10/pom.xml @@ -38,6 +38,9 @@ labx-10-sc-stream-rabbitmq-producer-transaction + + + labx-10-sc-stream-rabbitmq-consumer-ack