目录

基本概念

代码与实例

基本概念

首先有个关键&＃xff1a;此处实验接收的数据类型为Order&＃xff0c;这里要求发送和接收要一模一样。

包括包名和类名都要一模一样&＃xff1a;

如下&＃xff0c;consumerDemo

下面是productorDemo

这里&＃xff0c;包名和类都一模一样否则接收端监听会失败&＃xff01;

在消费&＃xff08;接收订阅&＃xff09;端要配置一些数据&＃xff1a;

spring.rabbitmq.listener.simple.concurrency&＃61;5
spring.rabbitmq.listener.simple.max-concurrency&＃61;10
spring.rabbitmq.listener.simple.acknowledge-mode&＃61;manual


这里指的是目前并发为5个&＃xff0c;最大并发数为10个&＃xff0c;监听确认为手动&＃xff0c;也就是接收了数据&＃xff0c;要给RabbitMQ给一个反馈信息

如下

这里有2个注解&＃xff0c;是简单使用RabbitMQ的关键&＃xff01;

&＃64;RabbitListener(bindings &＃61; &＃64;QueueBinding(value &＃61; &＃64;Queue(value &＃61; "order-queue", durable &＃61; "true"),exchange &＃61; &＃64;Exchange(name &＃61; "order-exchange", type &＃61; "topic"),key &＃61; "order.#"))&＃64;RabbitHandlerpublic void onOrderMessage(&＃64;Payload Order order,&＃64;Headers Map headers,Channel channel //手工签收需要rabbitMQ的通道) throws Exception{............
}

这里如果没有对应的交换机和队列&＃xff0c;那么此处就会自动新建

代码与实例

发送端不停的发送消息&＃xff01;接收端如下&＃xff1a;

发送端关键代码&＃xff1a;

OderSender.java

package SpringBoot.demo.produce;import SpringBoot.demo.entity.Order;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;&＃64;Component
public class OderSender {&＃64;Autowiredprivate RabbitTemplate rabbitTemplate;public void send(Order order) throws Exception{CorrelationData correlationData &＃61; new CorrelationData();correlationData.setId(order.getMessageId());rabbitTemplate.convertAndSend("order-exchange", //exchange"order.abcd", //routingKeyorder, //消息体correlationData); //correlationData消息唯一ID}
}


DemoApplicationTests.java

package SpringBoot.demo;import SpringBoot.demo.entity.Order;
import SpringBoot.demo.produce.OderSender;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;import java.util.UUID;&＃64;RunWith(SpringRunner.class)
&＃64;SpringBootTest
public class DemoApplicationTests {&＃64;Autowiredprivate OderSender oderSender;&＃64;Testpublic void contextLoads() {}&＃64;Testpublic void testSend1()throws Exception{Order order &＃61; new Order();order.setId("20180618000000000003");order.setName("测试订单3");order.setMessageId(System.currentTimeMillis() &＃43; "$" &＃43; UUID.randomUUID().toString());oderSender.send(order);}}


application.properties

server.port&＃61;8001
spring.rabbitmq.addresses&＃61;192.168.164.141:5672
spring.rabbitmq.username&＃61;guest
spring.rabbitmq.password&＃61;guest
spring.rabbitmq.virtual-host&＃61;/
spring.rabbitmq.connection-timeout&＃61;15000server.servlet.context-path&＃61;/
spring.http.encoding.charset&＃61;UTF-8
spring.jackson.data-format&＃61;yyyy-MM-dd HH:mm:ss
spring.jackson.time-zone&＃61;GMT&＃43;8
spring.jackson.default-property-inclusion&＃61;NON_NULL

接收端关键如下&＃xff1a;

OrderReceiver.java

package SpringBoot.demo.consumer;import SpringBoot.demo.entity.Order;
import com.rabbitmq.client.Channel;
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.util.Map;&＃64;Component
public class OrderReceiver {&＃64;RabbitListener(bindings &＃61; &＃64;QueueBinding(value &＃61; &＃64;Queue(value &＃61; "order-queue", durable &＃61; "true"),exchange &＃61; &＃64;Exchange(name &＃61; "order-exchange", type &＃61; "topic"),key &＃61; "order.#"))&＃64;RabbitHandlerpublic void onOrderMessage(&＃64;Payload Order order,&＃64;Headers Map headers,Channel channel //手工签收需要rabbitMQ的通道) throws Exception{//消费者操作System.out.println("--------------收到消息,开始消费--------------");System.out.println("订单ID&＃xff1a;" &＃43; order.getId());//告诉RabbitMQ我已经签收了long deliveryTag &＃61; (long)headers.get(AmqpHeaders.DELIVERY_TAG);channel.basicAck(deliveryTag, false); //false为不支持批量签收}}


application.properties

server.port&＃61;8002
spring.rabbitmq.addresses&＃61;192.168.164.141:5672
spring.rabbitmq.username&＃61;guest
spring.rabbitmq.password&＃61;guest
spring.rabbitmq.virtual-host&＃61;/
spring.rabbitmq.connection-timeout&＃61;15000server.servlet.context-path&＃61;/
spring.http.encoding.charset&＃61;UTF-8
spring.jackson.data-format&＃61;yyyy-MM-dd HH:mm:ss
spring.jackson.time-zone&＃61;GMT&＃43;8
spring.jackson.default-property-inclusion&＃61;NON_NULL#配置关于consumer相关的
spring.rabbitmq.listener.simple.concurrency&＃61;5
spring.rabbitmq.listener.simple.max-concurrency&＃61;10
spring.rabbitmq.listener.simple.acknowledge-mode&＃61;manual#限流&＃xff0c;同一时间只有一条消息消费
spring.rabbitmq.listener.simple.prefetch&＃61;1

源码下载地址&＃xff1a;

https://github.com/fengfanchen/Java/tree/master/ProductorDemo

https://github.com/fengfanchen/Java/tree/master/consumerDemo