Rabbit与springboot整合-1
目录
1、整体结构
2、pom引入
3、配置文件
4、代码
公共类
controller类
JSON转换类
监听-接收发送消息类
1、整体结构
2、pom引入
<!--rabbitmq--> <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId> </dependency>
3、配置文件
4、代码
公共类
controller类
import com.rabbitmqprovider.commons.CommonUtils;
import com.rabbitmqprovider.vo.User;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;@RestController
public class RabbitMQController {Logger LOG = LoggerFactory.getLogger(RabbitMQController.class);@AutowiredAmqpAdmin amqpAdmin;/*** 交换机创建*/@GetMapping("/createExchange")public void createExchange(){// 第一个参数为交换机名字，第二个参数为是否持久化，第三个参数为不使用交换机时删除DirectExchange directExchange = new DirectExchange(CommonUtils.dirExchange,true,false);amqpAdmin.declareExchange(directExchange);System.out.println("交换机创建成功");}/*** 绑定队列*/@GetMapping("/createQueue")public void createQueue() {/*** 第一个参数为队列名字，* 第二个参数为是否持久化，* 第三个参数为是否排他（true：一个连接只能有一个队列，false：一个连接可以有多个（推荐））* 第四个参数为不使用队列时自动删除*/Queue queue = new Queue(CommonUtils.queueStr,true,false,false);amqpAdmin.declareQueue(queue);System.out.println("队列创建成功");}/*** 绑定交换机和队列*/@GetMapping("/createBinding")public void createBinding() {/*** 第一个参数为目的地，就是交换机或者队列的名字* 第二个参数为目的地类型，交换机还是队列* 第三个参数为交换机，QUEUE-队列 EXCHANGE-交换机* 第四个参数为路由键，匹配的名称*/Binding binding = new Binding(CommonUtils.queueStr,Binding.DestinationType.QUEUE,CommonUtils.dirExchange,CommonUtils.routingKey,null);amqpAdmin.declareBinding(binding);System.out.println("绑定成功");}@AutowiredRabbitTemplate rabbitTemplate;/*** 发送消息* 结果:"这是一条消息"*/@GetMapping("/sendMessageTest")public String sendMessageTest(){// 消息类型为object 发送对象也是可以的String msg = "这是一条消息";// 第一个参数为发送消息到那个交换机上，第二个是发送的路由键（交换机进行需要符合绑定的队列），第三个参数为发送的消息rabbitTemplate.convertAndSend(CommonUtils.dirExchange,CommonUtils.routingKey,msg);System.out.println("消息发送成功");return "发送成功;发送内容为:"+msg;}/*** 发送消息* 结果：{"name":"张三、李四","age":22}*/@GetMapping("/sendMessageUserTest")public void sendMessageUserTest(){User user = new User();user.setAge(22);user.setName("张三、李四");// 第一个参数为发送消息到那个交换机上，第二个是发送的路由键（交换机进行需要符合绑定的队列），第三个参数为发送的消息rabbitTemplate.convertAndSend(CommonUtils.dirExchange,CommonUtils.routingKey,user);System.out.println("消息发送成功");}}
JSON转换类
监听-接收发送消息类
@Slf4j
@Service
public class TestService {// queues是监听的队列名字，可以是多个 ,CommonUtils.queueStrTest//接收全部的消息@RabbitListener(queues = {CommonUtils.queueStr})public void reciveMessage(Object message){System.out.println("接收的信息->" + message);}
返回结果:
接收的信息->(
Body:'"这是一条消息"' MessageProperties [headers={__TypeId__=java.lang.String}, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=helloword-java, receivedRoutingKey=hello.java, deliveryTag=1, consumerTag=amq.ctag-GWK_TTPoabVGD65ZtH81jg, consumerQueue=hello-java-queue])
//使用 @Payload注解 拿到 body的内容
@RabbitListener(queues = {CommonUtils.queueStr})
public void reciveMessage(@Payload String body, Channel channel,@Headers Map<String,Object> headers){System.out.println("接收的信息" + body);
}
返回结果:
接收的信息->这是一条消息
/*** 实体类接收* 接收的信息->张三、李四==22* @param body* @param channel*/
@RabbitListener(queues = {CommonUtils.queueStr})
public void reciveMessage(@Payload User body, Channel channel){System.out.println("接收的信息->" + body.getName()+"=="+body.getAge());
}
}
实体类
结果:
消息发送成功:这是一条消息
接收的信息(Body:'"这是一条消息"' MessageProperties [headers={__TypeId__=java.lang.String}, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=helloword-java, receivedRoutingKey=hello.java, deliveryTag=3, consumerTag=amq.ctag-cQAY1sFSowAWw2RH8IWuzA, consumerQueue=hello-java-queue])
提醒：刚才再运行项目，发送消息时，总是提示拒绝连接，研究了好长时间发现是端口错误了，我映射的端口是 5673->5672;