安装并启动RabbitMQ

1. 启动 RabbitMQ 服务器

./sbin/rabbitmq-server

Java 客户端访问

RabbitMQ 支持多种语言访问，以 Java 为例看下一般使用 RabbitMQ 的步骤。

1. maven工程的pom文件中添加依赖

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>4.1.0</version>
</dependency>

1. 消息生产者

package org.study.rabbitmq;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Producer {public static void main(String[] args) throws IOException, TimeoutException {//创建连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setUsername("guest");factory.setPassword("guest");//设置 RabbitMQ 地址factory.setHost("localhost");//建立到代理服务器到连接Connection conn = factory.newConnection();//获得信道Channel channel = conn.createChannel();//声明交换器String exchangeName = "hello-exchange";channel.exchangeDeclare(exchangeName, "direct", true);String routingKey = "hola";//发布消息byte[] messageBodyBytes = "quit".getBytes();channel.basicPublish(exchangeName, routingKey, null, messageBodyBytes);channel.close();conn.close();}
}

1. 消息消费者

package org.study.rabbitmq;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Consumer {public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();factory.setUsername("guest");factory.setPassword("guest");factory.setHost("localhost");//建立到代理服务器到连接Connection conn = factory.newConnection();//获得信道final Channel channel = conn.createChannel();//声明交换器String exchangeName = "hello-exchange";channel.exchangeDeclare(exchangeName, "direct", true);//声明队列String queueName = channel.queueDeclare().getQueue();String routingKey = "hola";//绑定队列，通过键 hola 将队列和交换器绑定起来channel.queueBind(queueName, exchangeName, routingKey);while(true) {//消费消息boolean autoAck = false;String consumerTag = "";channel.basicConsume(queueName, autoAck, consumerTag, new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body) throws IOException {String routingKey = envelope.getRoutingKey();String contentType = properties.getContentType();System.out.println("消费的路由键：" + routingKey);System.out.println("消费的内容类型：" + contentType);long deliveryTag = envelope.getDeliveryTag();//确认消息channel.basicAck(deliveryTag, false);System.out.println("消费的消息体内容：");String bodyStr = new String(body, "UTF-8");System.out.println(bodyStr);}});}}
}