RabbitMQ
Posted wakey
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了RabbitMQ相关的知识,希望对你有一定的参考价值。
RabbitMQ是用Erlang实现的一个高并发高可靠AMQP消息队列服务器。
支持消息的持久化、事务、拥塞控制、负载均衡等特性,使得RabbitMQ拥有更加广泛的应用场景。
使用场景:
场景1:单发送单接收
使用场景:简单的发送与接收,没有特别的处理。
package com.qa.util; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.concurrent.TimeoutException; public class ConnectionUtil { private static ConnectionFactory factory; static{ //连接rabbitmq服务器 factory = new ConnectionFactory(); factory.setHost("192.168.189.128"); factory.setPort(5672); factory.setUsername("admin"); factory.setPassword("admin"); factory.setVirtualHost("/admin"); } public static Connection getConnection(){ Connection connection = null; try { connection = factory.newConnection(); return connection; } catch (IOException e) { e.printStackTrace(); } catch (TimeoutException e) { e.printStackTrace(); } return null; } }
public class Provider { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); //获取通道Channel对象,后续所有的操作都是基于Channel实现 Channel channel = connection.createChannel(); //创建队列 channel.queueDeclare("queue1",false,false,false,null); //给队列发送消息 channel.basicPublish( "", "queue1", null, "Hello RabbitMQ".getBytes("utf-8")); //断开链接 connection.close(); } }
public class Consumer { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); //获取Channel对象 Channel channel = connection.createChannel(); //监听队列中的消息 channel.basicConsume("queue1",true,new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println("消费端获得消息:"+new String(body,"utf-8")); } }); } }
场景2:单发送多接收
使用场景:一个发送端,多个接收端,如分布式的任务派发。为了保证消息发送的可靠性,不丢失消息,使消息持久化了。同时为了防止接收端在处理消息时down掉,只有在消息处理完成后才发送ack消息。
public class Provider { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); //获取通道Channel对象,后续所有的操作都是基于Channel实现 Channel channel = connection.createChannel(); //创建队列 channel.queueDeclare("queue1",false,false,false,null); //给队列发送消息 for (int i=1;i<=10;i++){ channel.basicPublish( "", "queue1", null, ("Hello RabbitMQ"+i).getBytes("utf-8")); } //断开链接 connection.close(); } }
public class Consumer { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); //获取Channel对象 Channel channel = connection.createChannel(); //监听队列中的消息 channel.basicConsume("queue1",true,new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println("消费端1获得消息:"+new String(body,"utf-8")); } }); } }
public class Consumer2 { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); //获取Channel对象 Channel channel = connection.createChannel(); //监听队列中的消息 channel.basicConsume("queue1",true,new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println("消费端2获得消息:"+new String(body,"utf-8")); } }); } }
以上是关于RabbitMQ的主要内容,如果未能解决你的问题,请参考以下文章
带着新人学springboot的应用07(springboot+RabbitMQ 下)