大家好，又见面了，我是你们的朋友全栈君。
1.向application.properties文件中添加配置
#rabbitMQ的 5672 端口
spring.rabbitmq.addresses=192.168.31.199:32771
#用户名密码
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/
spring.rabbitmq.connection-timeout=15000
##springboot整合rabbitMQ 消费端配置
spring.rabbitmq.listener.simple.concurrency=5
spring.rabbitmq.listener.simple.max-concurrency=15
spring.rabbitmq.listener.simple.acknowledge-mode=manual
spring.rabbitmq.listener.simple.prefetch=1
2.创建消费者类
package com.youyou.consumer;
import com.rabbitmq.client.Channel;
import com.youyou.entity.Order;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.Map;
@Component
public class OrderReceive {
@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = "order-queue" ,durable = "true"),
exchange = @Exchange(name = "order-exchange" , durable = "true" , type = "topic"),
key = "order.#"
)
)
@RabbitHandler
public void onMessage(@Payload Order order , //
@Headers Map<String ,Object> headers ,
Channel channel) throws IOException {
//消费操作
System.out.println("接收到的订单：：" + order.getName());
Long tag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
//返回ack应答指令
channel.basicAck(tag,true);
}
}
启动项目之后，自动开始监听队列消息。
执行100万条数据效果如下：
没有丢失一条记录
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请联系我们举报，一经查实，本站将立刻删除。
发布者：全栈程序员-站长，转载请注明出处：https://javaforall.net/106469.html原文链接：https://javaforall.net
