RabbitMQ简单概念---打虎基本功
1.是什么
AMQP协议是一个高级抽象层消息通信协议，RabbitMQ是AMQP协议的实现,简单的来说它就是实现了:生产者生产了消息,将消息传递给消费者,然后消费者消费消息的过程
2.概念理解
生产者(Provider):
生产者其实是消息的提供者,一般用P(provider)表示,生产消息后将消息存储在队列
队列(Queue):
队列就是存储消息的中间仓库,生产者生产消息后存储在队列中,然后由消费者来消费
消费者(Consumer)
消费者故名思议就是从消息队列中取出消息进行消费的角色
示例场景---四面八方
第一篇HelloWorld示例采用的是RabbitMQ单发单收的场景实现,实现原理图如下:
图片.png
项目构建步骤---一步一个脚丫子
**1.创建项目 **
创建一个Maven项目,加入以下依赖,当然你也可以创建一个普通的项目,加入RabbitMQ的jar包,这里使用的版本是3.6.0
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>3.6.0</version>
</dependency>
2.实现生产者
package com.rabbitmq.helloworld;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class P {
private static final String QUEUE_NAME = "hello";
public static void main(String[] args) throws Exception {
//创建链接工厂
ConnectionFactory factory = new ConnectionFactory();
//设置rabbbitmq地址
factory.setHost("localhost");
//创建一个新的链接
Connection connection = factory.newConnection();
//创建一个频道
Channel channel = connection.createChannel();
//声明一个队列 -- 在RabbitMQ中，队列声明是幂等性的（一个幂等操作的特点是其任意多次执行所产生的影响均与一次执行的影响相同
//也就是说，如果不存在，就创建，如果存在，不会对已经存在的队列产生任何影响。
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
String message = "HELLO WORLD";
//发送消息到队列中
channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
System.out.println("P [x] Sent '" + message + "'");
//关闭频道和链接
channel.close();
connection.close();
}
}
3.实现消费者
package com.rabbitmq.helloworld;
import java.io.IOException;
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;
import com.rabbitmq.client.AMQP.BasicProperties;
public class C {
private static final String QUEUE_NAME = "hello";
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
//创建队列,为了保证启动的时候队列已经存在
channel.queueDeclare(QUEUE_NAME, false, false, false,null);
System.out.println("C [*] Waiting for messages. To exit press CTRL+C");
//DefaultConsumer类实现了Consumer接口，通过传入一个频道，告诉服务器我们需要那个频道的消息，如果频道中有消息，就会执行回调函数handleDelivery
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)throws IOException {
String message = new String(body, "UTF-8");
System.out.println("C [x] Received '" + message + "'");
}
};
//自动回复消息应答
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
运行测试---万无一失
1.开启RabbitMQ服务
我这里的安装路径是/usr/local/Cellar/rabbitmq/3.6.9/sbin/,根据你自己的输入
shizhengyangdeMacBook-Pro:~ panyuanyuan$ cd /usr/local/Cellar/rabbitmq/3.6.9/sbin/
shizhengyangdeMacBook-Pro:sbin panyuanyuan$ ./rabbitmq-server
RabbitMQ 3.6.9. Copyright (C) 2007-2016 Pivotal Software, Inc.
## ## Licensed under the MPL. See http://www.rabbitmq.com/
## ##
########## Logs: /usr/local/var/log/rabbitmq/rabbit@localhost.log
###### ## /usr/local/var/log/rabbitmq/rabbit@localhost-sasl.log
##########
Starting broker...
completed with 10 plugins.
2.运行客户端(消费者Consumer)
运行结果如下:
C [*] Waiting for messages. To exit press CTRL+C
3.运行生产者(Provider)
运行结果如下:
P [x] Sent 'HELLO WORLD'
4.观察消费者控制台
消费者控制台输出了消息,表示已经消费
C [*] Waiting for messages. To exit press CTRL+C
C [x] Received 'HELLO WORLD'
致谢
至此RabbitMQ的单发单收模式已经完成,感谢大大的微笑的博客,原博客链接 > http://blog.csdn.net/chwshuang/article/details/50521708
