1、安装RabbitMQ
安装好后,web页面访问:http://192.168.40.240:15672
2、添加依赖jar包
<dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit</artifactId> <version>1.4.6.RELEASE</version> </dependency> <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>2.2.0</version> </dependency>
3、生产者
public class Producter { private final static String QUEUE_NAME = "hello"; public static void main(String[] args) throws IOException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("192.168.40.240"); factory.setUsername("admin"); factory.setPassword("admin"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); String message = "Hello World!"; channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); channel.close(); connection.close(); } }
4、消费者
public class Consumer{ private final static String QUEUE_NAME = "hello"; public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setUsername("admin"); factory.setPassword("admin"); factory.setHost("192.168.40.240"); factory.setVirtualHost("/"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); QueueingConsumer consumer = new QueueingConsumer(channel); channel.basicConsume(QUEUE_NAME, true, consumer); while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [x] Received '" + message + "'"); } } }
5、启动生产者观察MQ管理页面
5、启动消费者观察MQ管理页面
demo2
public class Producter { public final static String QUEUE_NAME = "hello"; public static void main(String[] args) throws IOException, TimeoutException { //创建连接工厂 ConnectionFactory factory = new ConnectionFactory(); //设置RabbitMQ相关信息 factory.setHost("192.168.40.240"); factory.setPort(5672); factory.setUsername("admin"); factory.setPassword("admin"); //创建一个新的连接 Connection connection = factory.newConnection(); //创建一个通道 Channel channel = connection.createChannel(); // 声明一个队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); String message = "Hello RabbitMQ"; //发送消息到队列中 channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8")); System.out.println("Producer Send +'" + message + "'"); //关闭通道和连接 channel.close(); connection.close(); } }
public class Consumer{ private final static String QUEUE_NAME = "hello";//队列名称 private final static boolean DUEABLE = false;//消息持久化 private final static boolean ACK = false;//开启/关闭应答 public static void main(String[] args) throws IOException, TimeoutException { // 创建连接工厂 ConnectionFactory factory = new ConnectionFactory(); //设置RabbitMQ地址 factory.setHost("192.168.40.240"); factory.setPort(5672); factory.setUsername("admin"); factory.setPassword("admin"); //创建一个新的连接 Connection connection = factory.newConnection(); //创建一个通道 final Channel channel = connection.createChannel(); //每次接受一条消息 channel.basicQos(1); //声明要关注的队列 channel.queueDeclare(QUEUE_NAME, DUEABLE, false, false, null); System.out.println("Customer Waiting Received messages"); //DefaultConsumer类实现了Consumer接口,通过传入一个频道, //告诉服务器我们需要那个频道的消息,如果频道中有消息,就会执行回调函数handleDelivery DefaultConsumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String message = new String(body, "UTF-8"); System.out.println("Customer Received '" + message + "'"); // 处理业务 //应答 channel.basicAck(envelope.getDeliveryTag(), false); } }; //回复队列应答 -- RabbitMQ中的消息确认机制 channel.basicConsume(QUEUE_NAME, ACK, consumer); } }