上一章讲到rabbitmq的安装
这一章讲解rabbitmq的使用
首先启动rabbitmq
rabbitmq-server -detached
新建java springboot项目 这里我直接使用spring init
第一步：
pom文件中引入依赖
<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.8.0</version> </dependency>
第二步：创建consumer与producer端
消费端：
public class Consumer0 {/*** 创建交换机** 创建队列** 将队列与交换机绑定** 定义自己的消费方法**进行监听* @param args* @throws IOException* @throws TimeoutException*/public static void main(String[] args) throws IOException, TimeoutException {//队列的名字String queueName = "stu01queue";//交换机的名字String exchangeName = "stu01exchange";//1：创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setPort(5672);connectionFactory.setHost("192.168.106.70");connectionFactory.setVirtualHost("/");Connection connection = connectionFactory.newConnection();//2：获取channelChannel channel = connection.createChannel();/*** 3：定义exchange* 参数分别为：* exchange名字* exchange类型* 是否持久化* 自动删除？* 自定义参数*/channel.exchangeDeclare(exchangeName, "topic", true, true, null);/*** 4：* 定义queue** 参数分别为* 名字* 是否持久化* 是否顺序消费 只有我自己可以消费* 是否自动删除* 参数 拓展参数*/channel.queueDeclare(queueName, true, false, false, null);/*** 5：* 将队列绑定到exchange*/channel.queueBind(queueName, exchangeName, "stu0");//6：进行消费channel.basicConsume(queueName,true,new MyConsumer0(channel));}}
生产端：
public class Produce0 {public static String exchangeName = "stu01exchange";/*** 将消息发送到对应的交换机** 发送的时候指定routingkey** 这样 当消费者监听的队列绑定的是这个routingkey 就可以接受到消息*** @param args* @throws IOException* @throws TimeoutException*/public static void main(String[] args) throws IOException, TimeoutException {//工厂ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.106.70");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//链接Connection connection = connectionFactory.newConnection();//获取channelChannel channel = connection.createChannel();//创建要发送的数据String[] strs = {"迪丽热巴", "古力娜扎"};//将数组数据循环发送Arrays.stream(strs).forEach(e -> {try {channel.basicPublish(exchangeName, "stu0", null, e.getBytes());} catch (IOException ex) {ex.printStackTrace();}});String sendMsg = "Hello Rabbit";channel.basicPublish(exchangeName, "stu0", null, sendMsg.getBytes());channel.close();connection.close();}}
自定义消息接收方法：
public class MyConsumer0 extends DefaultConsumer {/*** Constructs a new instance and records its association to the passed-in channel.** @param channel the channel to which this consumer is attached*/public MyConsumer0(Channel channel) {super(channel);}@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("-----------consume message----------");System.out.println("consumerTag: " + consumerTag);System.out.println("envelope: " + envelope);System.out.println("properties: " + properties);System.out.println("body: " + new String(body));}
}
这样一个简单的topic类型的demo就完成了
启动consumer -> 启动producer
可以在控制台中看到已经成功了
在上述代码中 这里对一些值进行讲解：
durable：是否持久化 也就是是否将数据保存
autoDelete：是否自动删除 当与之绑定的队列被删除后 是否自动删除掉当前exchange？
argument：自定义的一些要发送的参数
其中有一个exchange类型是topic
其实还有direct 和fanout
topic模式下，代码中又一个绑定动作
topic对routingKey的绑定可以是一个模糊绑定 比如
test.# 意思是 test.（*）.(*) 多个值被路由到
test.* 代表 test.（*） 一个值被路由到
如果模式为direct 就是直连模式 只有routingKey刚好对应 才可以被路由
如果模式为fanout 就是类似于广播模式 所有监听exchange的都会被路由到
从上面不难看出 其实步骤就是三步
1:定义exchange
2定义queue
3将queue与exchange绑定起来
而在代码中 其实就是 消费端 监听一个queue
而生产端 将数据发送到一个exchange 并且指定发送时的routingkey
比如发送的routingkey 是 testRoutingKey
当消费端监听的队列与exchange绑定关系刚好是这个routingkey就会被路由
以上是简单的demo 后续有空我会将springboot中的使用简单方式进行详细的讲解