版权声明：本文为博主原创文章，如果转载请给出原文链接：http://doofuu.com/article/4156135.html
RabbitMQ简介
RabbitMQ是消费-生产者模型的一个典型的代表，一端往消息队列中不断写入消息，而另一端则可以读取或者订阅队列中的消息。它是一个实现了AMQP协议的可复用的企业消息系统。RabbitMQ是用Erlang开发。那么RabbitMQ也具备容错，热更新，分布式缓存，持久化等特点。RabbitMQ可以轻松的搭建出大型可用的分布式队列消息集群。
RabbitMQ的结构图：
RabbitMQ的一些概念：
Broker：简单来说就是消息队列服务器实体。
Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。
Queue：消息队列载体，每个消息都会被投入到一个或多个队列。
Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来。
Routing Key：路由关键字，exchange根据这个关键字进行消息投递。
vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离。
producer：消息生产者，就是投递消息的程序。
consumer：消息消费者，就是接受消息的程序。
channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务。
RabbitMQ是用Erlang开发的，但对于主要的编程语言都有驱动或者客户端。我们这里要用的是Java来作为客户端语言。假设都已经搭建好环境了（没有的话看这里），首先我们下载Java客户端类库，点击这里，下图是例子的模型：
首先创建队列，生产者向队列发送消息，消费者向队列取出消息，过程很简单，下面正式开始吧~~
步骤如下：
（一）创建访问用户
[root@localhost sbin]# ./rabbitmqctl add_user test test Creating user "test" ...
（二）给用户绑定vhost并添加权限
[root@localhost sbin]# ./rabbitmqctl set_permissions -p "/" test ".*" ".*" ".*" Setting permissions for user "test" in vhost "/" ...
这里就绑定默认的vhost“/”，也可以自己通过 ./rabbitmqctl add_vhost xx 来添加
（三）编写生产者程序并启动
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class HelloWorldProducer {
private final static String QUEUE_NAME = "hello";
public static void main(String[] args) throws Exception{
ConnectionFactory cf = new ConnectionFactory(); //rabbitmq监听IP
cf.setHost("192.168.1.96"); //rabbitmq默认监听端口，注意要记得开启端口
cf.setPort(5672);
//设置访问的用户
cf.setUsername("test");
cf.setPassword("test"); //建立连接
Connection conn = cf.newConnection(); //创建消息通道
Channel channel = conn.createChannel();
String msg = "hello world!!!! 你好啊~"; //创建hello队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//发送消息
channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
System.out.println("send msg "+ msg + " to ["+ QUEUE_NAME +"] queue !");
channel.close();
conn.close();
}
}可以看到hello队列里有一条消息了
[root@localhost sbin]# ./rabbitmqctl list_queues -p "/" Listing queues ... hello 1 [root@localhost sbin]#
（四）编写消费者程序并启动
import java.io.IOException;
import java.util.concurrent.TimeoutException;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
public class HelloWorldConsumer {
private final static String QUEUE_NAME = "hello";
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory cf = new ConnectionFactory(); //rabbitmq监听IP
cf.setHost("192.168.1.96"); //rabbitmq默认监听端口，注意要记得开启端口
cf.setPort(5672);
//设置访问的用户
cf.setUsername("test");
cf.setPassword("test"); //建立连接
Connection conn = cf.newConnection(); //创建消息通道
Channel channel = conn.createChannel(); //创建hello队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
System.out.println(" Waiting for msg...."); //创建消费者，并接受消息
Consumer consumer = new DefaultConsumer(channel) { @Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
throws IOException {
String msg = new String(body, "UTF-8");
System.out.println("Received is = '" + msg + "'");
}
};
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}运行结果：
这时队列里的消息被消费掉了
[root@localhost sbin]# ./rabbitmqctl list_queues -p "/" Listing queues ... hello 0 [root@localhost sbin]#
如果不成功可能是一下几点造成的：
如果连接不成功可能是IP不对或者是5672端口没开启，可以在rabbitmq.config中修改(默认是{tcp_listeners, [{"127.0.0.1",5672}]})
如果用guest(默认用户)访问，监听Host则必须是localhost
如果是新创建的用户，记得给用户绑定vhost并添加权限
可以从RabbitMQ的日志文件 rabbit@localhost.log ，获取更多详细的错误信息
共有 0 条评论 - RabbitMQ之入门HelloWorld(Java)