这篇文章通过一个最简单的例子,让初学者能了解RabbitMQ如何完成生产消息和消息的。
所有的程序员在学习一门新技术的时候,都是从 Hello World 进入到Colorful World的,本节也将按照惯例,从HelloWorld开始,演示RabbitMQ的Produce和Consumer的简单使用。本RabbitMQ系列的演示代码默认都是使用Java语言。
设置账号
在开始HelloWorld之前,需要注意的是,RabbitMQ默认的账号是guest / guest,这个账号有限制,默认只能通过本地网络(localhost)访问,远程访问受限制,所以在实际发送和消费消息之前,需要设置新的账号和设置权限。具体账号和权限的内容敬请关注后面的更新。
添加账号
我们为HelloWorld创建一个新的用户为root,并设置密码为root,后续Java客户端代码中使用这个root账号发送和消费消息。
[root@hidden -]# rabbitmqct1 add user root root
Creating user "root"
设置权限
在创建好账号之后就要为这个账号创建权限了。
[root@hidden - ]# rabbitmqct1 set_permissions -p / root ".*" ".*" ".*"
Setting permissions for user "root" in vhost "/"
设置角色
最后需要为这个账号添加角色,这里我们添加管理员角色
[root@hidden - ]# rabbitmqct1 set user_tags root administrator
Setting tags for user "root" to [administrator]
经过上面的步骤,root账号就已经创建成功,也可以通过客户端链接rabbitmq的broker,如果遇到下面的问题,说明是账号出现的问题,参考上面的步骤设置,或者检查账号是否正确。
Exception in thread "main" com.rabbitmq.c1ient.AuthenticationFai1ureException :
ACCESS REFUSED - Login was refused using authentication mechanism PLAIN. For details see the broker 1ogfi1e.
添加Maven依赖
RabbitMQ的java版客户端的maven以来如下,可以根据自己的环境选择具体的版本即可。
<!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
<dependency><groupld>com.rabbitmq</groupld><artifactld>amqp-client</artifactld><version>${rabbitmq.version}</version>
</dependency>
Producer案例
Producer就是用来向MQ发送消息,下面就是一个Producer的HelloWorld。
import com.rabbitmq.client.*;
import org.junit.jupiter.api.Test;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class ProducerTest {@Testpublic void helloWorld() throws IOException, TimeoutException {// 交换器名字String exchangeName = "helloworld_exchange";// 路由键String routingKey = "helloworld_routing_key";// 队列名字String queueName = "helloworld_queue";// 创建连接工厂,用来创建具体的连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("localhost");connectionFactory.setPort(5672);connectionFactory.setUsername("root");connectionFactory.setPassword("root");// 根据上面的设置信息,创建具体的连接Connection connection = connectionFactory.newConnection();// 在创建的连接上,创建一个通道Channel channel = connection.createChannel();// 在通道上声明交换器channel.exchangeDeclare(exchangeName, "direct", true, false, null);// 在通道上声明队列channel.queueDeclare(queueName, true, false, false, null);// 声明交换器和队列的绑定关系channel.queueBind(queueName, exchangeName, routingKey);String message = "hello world";// 往通道上发送消息,消息通过绑定键发送到指定的队列,也就是上面申明的绑定关系的队列channel.basicPublish(exchangeName, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());// 关闭通过和连接channel.close();connection.close();}}
Consumer案例
消息发送到MQ,Consumer就可以订阅队列,并开始消费消息,下面就是一个Consumer的HelloWorld。
import com.rabbitmq.client.*;
import org.junit.jupiter.api.Test;import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;public class ConsumerTest {@Testpublic void helloWorld() throws IOException, TimeoutException, InterruptedException {// 订阅的队列String queueName = "helloworld_queue";// 订阅队列所在的broker的地址信息,这里演示另一种创建连接的方式Address[] addresses = {new Address("localhost", 5672)};ConnectionFactory factory = new ConnectionFactory();factory.setUsername("root");factory.setPassword("root");Connection connection = factory.newConnection(addresses);Channel channel = connection.createChannel();// consumer端一般需要设置的值,表示一次消费的消息数量的最大值channel.basicQos(64);// 创建默认的Consumer,并实现回调函数,在回调中确认消息,发送ackConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println(Thread.currentThread().getName() + " -> consumer : " + consumerTag + " , receive message : " + new String(body));channel.basicAck(envelope.getDeliveryTag(), false);}};// 消费消息,异步执行String consumeTag = channel.basicConsume(queueName, consumer);System.out.println(consumeTag);// 等到消费消息完成,并返回确认给brokerTimeUnit.SECONDS.sleep(10);// 关闭资源channel.close();connection.close();}
}
以上就是通过一个HelloWorld,了解下RabbitMQ的简单使用,后续会不定期更新RabbitMQ的内容,感兴趣的小伙伴敬请关注哦。