rabbitmq快速上手

rabbitmq快速上手

1. docker拉取rabbitmq容器

docker run -d
–hostname my-rabbit
–name some-rabbit
-e RABBITMQ_DEFAULT_USER=user
-e RABBITMQ_DEFAULT_PASS=password
-p 15672:15672 -p 5672:5672
rabbitmq:3-management

---

– hostname:在配置集群时使用，区分节点。(这里是单机，随便起个名字)
-p 15672 rabbitmq 提供的可视化平台的端口号，便于管理
-p 5672 rabbitmq 服务器的端口，用于消息的发布和订阅
RABBITMQ_DEFAULT_USER、RABBITMQ_DEFAULT_PASS：用来配置登录可视化管理平台的账号，密码

2. 开放端口号：
   firewall-cmd --add-port=15672/tcp --permanent
   firewall-cmd --add-port=5672/tcp --permanent
   firewall-cmd --reload 加载开放的端口号
3. 访问rabbitmq的管理平台：
   ip:15672
   使用刚刚设置的用户名，密码登录
4. 添加一个虚拟主机（一个虚拟机下有一套自己的交换机和队列，便于不同用户管理）
5. 下载演示以下三种交换机的使用方式(fanout、direct、topic)
   1. 创建一个父项目：rabbitmq
      父项目下创建两个模块：publisher、consumer
      在父项目中引入依赖：

		<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.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId></dependency>

6. 添加一个Been，防止发送消息(对象) 时乱码，发布消息默认采用的jdk的object序列化，添加一个json的消息转换器。publisher、consumer模块都要用到（需要序列化和反序列化）

	@Beanpublic MessageConverter messageConverter(){return new Jackson2JsonMessageConverter();}

7. 配置文件（publisher、consumer都需要，配置不同的server.port）

server:port: 8080
spring:rabbitmq:host: 192.168.193.105 port: 5672username: wuyupassword: 123456virtual-host: "/wuyu" # 自己创建的虚拟主机

8. consumer中创建消息监听器
   下面是三种模式 exchange的监听方式

package sumer.listener;import org.ExchangeTypes;
import org.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
public class MessageQueueListener {@RabbitListener(bindings = {@QueueBinding(value = @Queue("fanout.queue1"), exchange = @Exchange(name = "wuyu.fanout", type = ExchangeTypes.FANOUT))})public void fanoutQueue1(Message msg) {byte[] body = Body();String res = new String(body);System.out.println("fanout.queue1：" + res);}@RabbitListener(bindings = {@QueueBinding(value = @Queue("fanout.queue2"), exchange = @Exchange(name = "wuyu.fanout", type = ExchangeTypes.FANOUT))})public void fanoutQueue2(Message msg) {byte[] body = Body();String res = new String(body);System.out.println("fanout.queue2：" + res);}@RabbitListener(bindings = {@QueueBinding(value = @Queue("direct.queue1"),exchange = @Exchange(name = "wuyu.direct", type = ExchangeTypes.DIRECT),key = {"red"})})public void directQueue1(Message msg) {byte[] body = Body();String res = new String(body);System.out.println("direct.queue1：" + res);}@RabbitListener(bindings = {@QueueBinding(value = @Queue("direct.queue2"),exchange = @Exchange(name = "wuyu.direct", type = ExchangeTypes.DIRECT),key = {"yellow"})})public void directQueue2(Message msg) {byte[] body = Body();String res = new String(body);System.out.println("direct.queue2：" + res);}@RabbitListener(bindings = {@QueueBinding(value = @Queue("topic.queue1"),exchange = @Exchange(name = &#pic", type = ExchangeTypes.TOPIC),key = {"china.#"})})public void topicQueue1(Message msg) {byte[] body = Body();String res = new String(body);System.out.println("topic.queue1：" + res);}@RabbitListener(bindings = {@QueueBinding(value = @Queue("topic.queue2"),exchange = @Exchange(name = &#pic", type = ExchangeTypes.TOPIC),key = {"#.news"})})public void topicQueue2(Message msg) {byte[] body = Body();String res = new String(body);System.out.println("topic.queue2：" + res);}
}

9. publisher 中创建测试类，验证三种exchange

@SpringBootTest
public class PublisherAppTest {@Resourceprivate RabbitTemplate rabbitTemplate;/*** fanout模式，不需要 routingKey。* 它会把消息转发到每一个和 fanout交换机绑定的队列*/@Testpublic void testSendFanoutExchange(){String exchange="wuyu.fanout";Map<String,String> map=new HashMap<>();map.put("name","小舞");vertAndSend(exchange,"",map);}/*** direct模式，需要 routingKey* 它会把消息转发到 指定 routingKey的队列*/@Testpublic void testDirectExchange(){String exchange="wuyu.direct";String routingKey="red";Map<String,String> map=new HashMap<>();map.put("name","小舞");vertAndSend(exchange,routingKey,map);}/*** topic模式，需要 routingKey (在direct的基础上 routingKey可以使用通配符)* ( # 匹配 0或多个单词 )* ( * 匹配一个单词 )* 匹配到 routingKey 的监听队列会接收发布的消息*/@Testpublic void testTopicExchange(){String exchange=&#pic";String routingKey="china.video";Map<String,String> map=new HashMap<>();map.put("name","小舞");vertAndSend(exchange,routingKey,map);}
}