赞
踩
RabbitMQ是一个开源的消息代理和队列服务器,用来通过普通协议在完成不同的应用之间共享数据,RabbitMQ是使用Erlang语言来编写的,并且RabbitMQ是基于AMQP协议的。
AMQP全称:Advanced Message Queuing Protocol
AMQP翻译:高级消息队列协议
docker 拉取rabbitmq(默认最新)镜像
docker pull rabbitmq
docker运行rabbitmq
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 -v /mydata/rabbitmq/data:/var/lib/rabbitmq --hostname myRabbit -e RABBITMQ_DEFAULT_VHOST=my_vhost -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin 586ccc4c0d63(镜像编号)
开启web控制台
# 进入rabbitmq容器
[vagrant@bogon etc]$ sudo docker exec -it rabbitmq /bin/bash
# 进入rabbitmq容器,开启web控制台
root@myRabbit:/# rabbitmq-plugins enable rabbitmq_management
进入rabbitmq容器,执行如下命令。
cd /etc/rabbitmq/conf.d/
echo management_agent.disable_metrics_collector = false > management_agent.disable_metrics_collector.conf
exit
在rabbitMq中,Exchange有四个类型:direct,topic,fanout,header,详见
direct类型的exchange,只有当routing key完全相等时,才能路由到对应的Queue
topic类型的direct差不多,只是topic可以设置“*”和“#”通配符。
俗称广播,此路由直接将消息路由到所有绑定队列中,无须对消息的routingkey进行匹配。
Headr类型的exchange是根据header来判断,用的比较少
此节简单介绍一下,springboot如何使用rabbitmq,在此之前,我们需要两个springboot工程(生产者和消费者),他们的包名结构一定要相同,因为这涉及到消费者接收消息时,如果两个工程的消息类结构和包名不一样会接收不到。来看一下我的包结构,依次是生产者,消费者
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.6.1</version> <relativePath/> <!-- lookup parent from repository --> </parent> <groupId>com.study.rabbitmq</groupId> <artifactId>RApusic-consumer</artifactId> <version>0.0.1-SNAPSHOT</version> <name>RApusic-consumer</name> <description>study for Spring Boot</description> <properties> <java.version>1.8</java.version> </properties> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <optional>true</optional> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit-test</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-lang3</artifactId> </dependency> <dependency> <groupId>commons-io</groupId> <artifactId>commons-io</artifactId> <version>2.6</version> </dependency> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>1.2.78</version> </dependency> <dependency> <groupId>javax.servlet</groupId> <artifactId>javax.servlet-api</artifactId> <scope>provided</scope> </dependency> <dependency> <groupId>log4j</groupId> <artifactId>log4j</artifactId> <version>1.2.17</version> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-maven-plugin</artifactId> <configuration> <excludes> <exclude> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> </exclude> </excludes> </configuration> </plugin> </plugins> </build> </project>
spring: http: encoding: charset: UTF-8 datasource: url: jdbc:mysql://localhost:3306/ml?useUnicode=true&characterEncoding=UTF-8&allowMultiQueries=true&serverTimezone=GMT%2B8 username: root password: root jackson: date-format: yyyy-MM-dd HH:mm:ss time-zone: GMT+8 default-property-inclusion: NON_NULL rabbitmq: addresses: 192.168.137.20:5672 #rabbitmq地址 username: admin password: admin virtual-host: / #虚拟主机,一般最新版好像没有这个虚拟主机,需要你去rabbitmq的web控制台创建一下。 connection-timeout: 15000 server: servlet: context-path: / port: 8001
写完配置文件后,就该开始编写代码了,首先编写实体类 Order。
package com.study.rabbitmq.entity; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; import java.io.Serializable; @Data @NoArgsConstructor @AllArgsConstructor public class Order implements Serializable { private static final long serialVersionUID = -837945093037776296L; private String id; private String name; private String messageId;//存储消息发送的唯一标识。 }
生产者发送消息的代码。由于springboot的关系,所以写的代码非常少
package com.study.rabbitmq.producer; import com.study.rabbitmq.entity.Order; import org.springframework.amqp.rabbit.connection.CorrelationData; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @Component public class OrderSender { @Autowired private RabbitTemplate rabbitTemplate; public void send(Order order){ CorrelationData correlationData = new CorrelationData(); correlationData.setId(order.getMessageId()); rabbitTemplate.convertAndSend("order-exchange",//exchange "order.dada.123",//routingKey order,//消息体内容 correlationData);//correlationData消息唯一ID } }
到此生产者代码编写完毕,是不是很简单。但是完成这些代码后我们发的消息,还不能被rabbitmq接收,因为exchange和queue还没有在rabbit中创建。所以下一节的内容,聪明的你应该猜到了。
输入用户密码登录控制台。
点击exchange选项卡,点击add a new queue
点开后,键入name(order-queue),点击Add queue
点exchange选项卡,点击创建好的order-exchange
键入图中内容,点击bind按钮,完成绑定。
到此rabbitmq的exchange和queue设置完毕。我们可以测试一下,是否可以使用(ps:如果控制台点击遇到Management API returned status code 500的错误,请看安装那节遇到问题篇)。
在生产者中编写测试用例并启动。
package com.study.rabbitmq.RApusic; import com.study.rabbitmq.entity.Order; import com.study.rabbitmq.producer.OrderSender; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import java.util.UUID; @SpringBootTest class RApusicApplicationTests { @Test void contextLoads() { System.out.printf("111"); } @Autowired private OrderSender orderSender; @Test public void testSend(){ Order order = new Order(); order.setId("2021120900000001"); order.setName("测试订单1"); order.setMessageId(System.currentTimeMillis()+"$"+ UUID.randomUUID().toString()); orderSender.send(order); } }
查看web控制台,可见消息已经发送到了rabbitmq中。
生产者生产的消息,需要消费者消费。
spring: http: encoding: charset: UTF-8 datasource: url: jdbc:mysql://localhost:3306/ml?useUnicode=true&characterEncoding=UTF-8&allowMultiQueries=true&serverTimezone=GMT%2B8 username: root password: root jackson: date-format: yyyy-MM-dd HH:mm:ss time-zone: GMT+8 default-property-inclusion: NON_NULL rabbitmq: #基本配置 addresses: 192.168.137.20:5672 username: admin password: admin virtual-host: / connection-timeout: 15000 #消费端配置 listener: simple: concurrency: 5 #并发数 max-concurrency: 15 #最大并发数 acknowledge-mode: manual #签收模式 manual(手动)or AUTO(自动) prefetch: 1 #限流 server: servlet: context-path: / port: 8002
消费者也需要entity来接收发送来的消息。
package com.study.rabbitmq.entity; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; import java.io.Serializable; @Data @NoArgsConstructor @AllArgsConstructor public class Order implements Serializable { private static final long serialVersionUID = -837945093037776296L; private String id; private String name; private String messageId;//存储消息发送的唯一标识。 }
编写监听
package com.study.rabbitmq.consumer; import com.rabbitmq.client.Channel; import com.study.rabbitmq.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 OrderReceiver { @RabbitListener(bindings = @QueueBinding(value = @Queue(value = "order-queue",declare = "true"), exchange = @Exchange(name = "order-exchange",durable = "true",type ="topic" ), key = "order.#" )) @RabbitHandler public void onOrderMessage(@Payload Order order, @Headers Map<String,Object> headers, Channel channel) throws IOException { //消息者操作 System.out.println("-----收到消息------"); System.out.println("订单ID"+order.getId()); //获取heads中的DELIVERY_TAG Long deliveryTag =(Long) headers.get(AmqpHeaders.DELIVERY_TAG); //手工签收,basicAck(DELIVERY_TAG,是否支持批量); channel.basicAck(deliveryTag,false); } }
启动springboot,
至此springboot整合rabbitmq完毕。如果您发现文章中有错误,请及时告诉我,万分感谢。
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。