（转） RabbitMQ学习之helloword（java）

2023-08-30 10:18:46

http://blog.csdn.net/zhu_tianwei/article/details/40835555

amqp-client:http://www.rabbitmq.com/java-client.html

1.依赖jar包

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

2.生产者代码Send.Java

  1. package cn.slimsmart.rabbitmq.demo.helloword;
  2. import com.rabbitmq.client.AMQP;
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  6. public class Send {
  7. //消息队列名称
  8. private final static String QUEUE_NAME = "helloword";
  9. public static void main(String[] args) throws Exception {
  10. /**
  11. * 创建连接连接到MabbitMQ
  12. */
  13. ConnectionFactory factory = new ConnectionFactory();
  14. //设置MabbitMQ所在主机ip或者主机名
  15. factory.setHost("192.168.101.174");
  16. //指定用户 密码
  17. factory.setUsername("admin");
  18. factory.setPassword("admin");
  19. //指定端口
  20. factory.setPort(AMQP.PROTOCOL.PORT);
  21. //创建一个连接
  22. Connection connection = factory.newConnection();
  23. //创建一个频道
  24. Channel channel = connection.createChannel();
  25. //指定一个队列
  26. channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  27. //发送的消息
  28. String message = "hello world!";
  29. //往队列中发出一条消息
  30. channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
  31. System.out.println("Sent Message：'" + message + "'");
  32. //关闭频道和连接
  33. channel.close();
  34. connection.close();
  35. }
  36. }

3.消费者代码Receive.java

  1. package cn.slimsmart.rabbitmq.demo.helloword;
  2. import com.rabbitmq.client.AMQP;
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  6. import com.rabbitmq.client.QueueingConsumer;
  7. public class Receive {
  8. //消息队列名称
  9. private final static String QUEUE_NAME = "helloword";
  10. public static void main(String[] args) throws Exception {
  11. //打开连接和创建频道，与发送端一样
  12. ConnectionFactory factory = new ConnectionFactory();
  13. factory.setHost("192.168.101.174");
  14. //指定用户 密码
  15. factory.setUsername("admin");
  16. factory.setPassword("admin");
  17. //指定端口
  18. factory.setPort(AMQP.PROTOCOL.PORT);
  19. //创建一个连接
  20. Connection connection = factory.newConnection();
  21. //创建一个频道
  22. Channel channel = connection.createChannel();
  23. //声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。
  24. channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  25. //创建队列消费者
  26. QueueingConsumer consumer = new QueueingConsumer(channel);
  27. //指定消费队列
  28. channel.basicConsume(QUEUE_NAME, true, consumer);
  29. while (true)
  30. {
  31. //nextDelivery是一个阻塞方法（内部实现其实是阻塞队列的take方法）
  32. QueueingConsumer.Delivery delivery = consumer.nextDelivery();
  33. String message = new String(delivery.getBody());
  34. System.out.println("Received Message：'" + message + "'");
  35. }
  36. }
  37. }

如果运行出现如下异常，可能创建的用户没有访问权限。

  1. Exception in thread "main" java.io.IOException
  2. at com.rabbitmq.client.impl.AMQChannel.wrap(AMQChannel.java:106)
  3. at com.rabbitmq.client.impl.AMQChannel.wrap(AMQChannel.java:102)
  4. at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:124)
  5. at com.rabbitmq.client.impl.AMQConnection.start(AMQConnection.java:388)
  6. at com.rabbitmq.client.ConnectionFactory.newConnection(ConnectionFactory.java:516)
  7. at com.rabbitmq.client.ConnectionFactory.newConnection(ConnectionFactory.java:533)
  8. at cn.slimsmart.rabbitmq.demo.test.Test.main(Test.java:18)
  9. Caused by: com.rabbitmq.client.ShutdownSignalException: connection error; reason: java.net.SocketException: Connection reset
  10. at com.rabbitmq.utility.ValueOrException.getValue(ValueOrException.java:67)
  11. at com.rabbitmq.utility.BlockingValueOrException.uninterruptibleGetValue(BlockingValueOrException.java:33)
  12. at com.rabbitmq.client.impl.AMQChannel$BlockingRpcContinuation.getReply(AMQChannel.java:343)
  13. at com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:216)
  14. at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:118)
  15. ... 4 more
  16. Caused by: java.net.SocketException: Connection reset
  17. at java.net.SocketInputStream.read(Unknown Source)
  18. at java.net.SocketInputStream.read(Unknown Source)
  19. at java.io.BufferedInputStream.fill(Unknown Source)
  20. at java.io.BufferedInputStream.read(Unknown Source)
  21. at java.io.DataInputStream.readUnsignedByte(Unknown Source)
  22. at com.rabbitmq.client.impl.Frame.readFrom(Frame.java:95)
  23. at com.rabbitmq.client.impl.SocketFrameHandler.readFrame(SocketFrameHandler.java:131)
  24. at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:515)

需要对该用户进行授权，登录web控制台后，点击"admin"，进入需要授权的用户，在Admin标签页下点击新增的用户"admin"，进入授权页面，默认直接点击"set permission"即可

（转） RabbitMQ学习之helloword（java）

实例代码：http://download.csdn.net/detail/tianwei7518/8136413

上一篇：linux下安装rabbitmq 集群


下一篇：RabbitMQ学习之集群部署