上海网站建设 分类广告公司网站设计

               目录

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;