直销系统网站开发/晋城seo
概念
rabbitmq概念比较多,这里说几个重要的
routing key: 消息体的一部分 发送消息时指定,用于队列和交换机绑定关系进行匹配。
exchange 交换机:rabbitmq发送消息直接发送到交换机,由交换机发送消息给队列。
queue:用于存储消息,交换机发送的消息会进入这里面,在创建的时候可以指定消息持久化,设置durable属性为TRUE。
bingKey 用于绑定交换机和队列的关系,在项目初始化的时候声明,bindkey决定了通配符的消息进入哪些队列
通配符:支持 *和# 比如 a.b.c 就是三个单词,单词多少以点为界限确定 #可以匹配多个单词*只可以匹配一个单词
简单结构图如下:
发送方式
rabbitmq支持了点对点,通配符,广播模式等发送方式。
下面主要就这三点做一个简单概述,示例代码使用springboot
点对点
发送消息指定消息routing key 为 a.b ,只有接收端交换机和队列使用a.b绑定的时候才可以使用
配置类
package com.tom.config;import com.tom.message.Demo01Message;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitConfig {/*** Direct Exchange 示例的配置类*/public static class DirectExchangeDemoConfiguration {// 创建 Queue//点对点发布@Beanpublic Queue demo01Queue() {return new Queue(Demo01Message.QUEUE, // Queue 名字true, // durable: 是否持久化false, // exclusive: 是否排它false); // autoDelete: 是否自动删除}// 创建 Direct Exchange@Beanpublic DirectExchange demo01Exchange() {return new DirectExchange(Demo01Message.EXCHANGE,true, // durable: 是否持久化false); // exclusive: 是否排它}// 创建 Binding// Exchange:Demo01Message.EXCHANGE// Routing key:Demo01Message.ROUTING_KEY// Queue:Demo01Message.QUEUE@Beanpublic Binding demo01Binding() {return BindingBuilder.bind(demo01Queue()).to(demo01Exchange()).with(Demo01Message.ROUTING_KEY);}}}
消费者:
package com.tom.mq;// Demo01Consumer.javaimport com.tom.message.Demo01Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
@RabbitListener(queues = Demo01Message.QUEUE,concurrency = "5")
public class Demo01Consumer {private Logger logger = LoggerFactory.getLogger(getClass());@RabbitHandlerpublic void onMessage(Demo01Message message) {logger.info("[onMessage][线程编号:{} 消息内容:{}]", Thread.currentThread().getId(), message);}}
发送端:
rabbitTemplate.convertAndSend(Demo01Message.EXCHANGE, Demo01Message.ROUTING_KEY, message);
topic方式发送
配置类,在点对点的配置类的基础上加上内部类
/*** 下面的配置为通配符发布 topic发布*/public static class LogExchangeConfiguration{// 创建 Queue@Beanpublic Queue logQueue1() {return new Queue(Constant.myQueue1, // Queue 名字true, // durable: 是否持久化false, // exclusive: 是否排它false); // autoDelete: 是否自动删除}@Beanpublic Queue logQueue2() {return new Queue(Constant.myQueue2, // Queue 名字true, // durable: 是否持久化false, // exclusive: 是否排它false); // autoDelete: 是否自动删除}// 创建 Topic Exchange@Beanpublic TopicExchange logExchange() {return new TopicExchange(Constant.myExchange,true, // durable: 是否持久化false); // exclusive: 是否排它}// 创建 Binding//一个queue 多个@Beanpublic Binding BindingLogQueue1() {return BindingBuilder.bind(logQueue1()).to(logExchange()).with("a.b");}@Beanpublic Binding bindingLogQueue2() {return BindingBuilder.bind(logQueue2()).to(logExchange()).with("a.*");}}
消费者:
package com.tom.mq;// Demo01Consumer.javaimport com.tom.config.Constant;
import com.tom.message.Demo01Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
@RabbitListener(queues = Constant.myQueue1,concurrency = "5")
public class Consumer1 {private Logger logger = LoggerFactory.getLogger(getClass());@RabbitHandlerpublic void onMessage( String message) {logger.info("myQueue1收到消息【{}】", message);}}
package com.tom.mq;// Demo01Consumer.javaimport com.tom.config.Constant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
@RabbitListener(queues = Constant.myQueue2,concurrency = "5")
public class Consumer2 {private Logger logger = LoggerFactory.getLogger(getClass());@RabbitHandlerpublic void onMessage( String message) {logger.info("myQueue2收到消息{}", message);}}
发送端:
package com.tom;import com.tom.config.Constant;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class TopicTest {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void test() {String message="hello a--------------------->>>>>>>>>>>>>>>>>>>>>";rabbitTemplate.convertAndSend(Constant.myExchange,"a.b",message);System.out.println("发送消息success!!!!");}
}
发送结果:
发布模式和上面两种类似,这里暂时不做赘述。
开发总结
1.声明队列,交换机,建立交换机和队列的绑定关系
2.建立监听队列的监听器
3.发送消息指定消息的routingkey
4.接收消息处理业务
代码地址:
https://github.com/weiwensi/cloud-parent.git
地址上项目的rabbitmq模块