rabbitmq学习1:hello world _JAVA_编程开发_程序员俱乐部

rabbitMQ是一个在AMQP基础上完整的，可服用的企业消息系统。他遵循Mozilla Public License 开源协议。

? 关于amqp可参考http://www.oschina.net/p/rabbitmq/

?rabbitmq是一个消费的代理；通过生产者客户端生产一个信息，转送给消费者客户端;在这个传输过程中，根据你的需要可以经过路由、缓冲、持久化来得到这个消息。

? 先通过一个例子开始：通过rabbitmq输出"Hello World!"

?

?

其中P代表生产者、C表示消费者、中间红色部分代表消息队列

?

生产者客户端的发送消息程序如下：

?

1. package?com.abin.test; ??
2. ??
3. import?java.io.IOException; ??
4. ??
5. import?com.rabbitmq.client.Channel; ??
6. import?com.rabbitmq.client.Connection; ??
7. import?com.rabbitmq.client.ConnectionFactory; ??
8. ??
9. public?class?Send?{ ??
10. ????private?final?static?String?QUEUE_NAME?=?"hello"; ??
11. ??
12. ????public?static?void?main(String[]?args)?throws?IOException?{ ??
13. ????????ConnectionFactory?factory?=?new?ConnectionFactory(); ??
14. ????????factory.setHost("localhost"); ??
15. ????????Connection?connection?=?factory.newConnection(); ??
16. ????????Channel?channel?=?connection.createChannel(); ??
17. ??
18. ????????channel.queueDeclare(QUEUE_NAME,?false,?false,?false,?null); ??
19. ????????String?message?=?"Hello?World!"; ??
20. ????????channel.basicPublish("",?QUEUE_NAME,?null,?message.getBytes()); ??
21. ????????System.out.println("?[x]?Sent?'"?+?message?+?"'"); ??
22. ??
23. ????????channel.close(); ??
24. ????????connection.close(); ??
25. ????} ??
26. }??

package com.abin.test;

import java.io.IOException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class Send {
	private final static String QUEUE_NAME = "hello";

	public static void main(String[] args) throws IOException {
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("localhost");
		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();

		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
		String message = "Hello World!";
		channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
		System.out.println(" [x] Sent '" + message + "'");

		channel.close();
		connection.close();
	}
}

运行结果如下：

1. [x]?Sent?'Hello?World!'??

 [x] Sent 'Hello World!'

?

消费者客户端接收消息程序如下：

?

1. package?com.abin.test; ??
2. ??
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. ??
8. public?class?Reqv?{ ??
9. ????private?final?static?String?QUEUE_NAME?=?"hello"; ??
10. ??
11. ????public?static?void?main(String[]?argv)?throws?Exception?{ ??
12. ??
13. ????????ConnectionFactory?factory?=?new?ConnectionFactory(); ??
14. ????????factory.setHost("localhost"); ??
15. ????????Connection?connection?=?factory.newConnection(); ??
16. ????????Channel?channel?=?connection.createChannel(); ??
17. ??
18. ????????channel.queueDeclare(QUEUE_NAME,?false,?false,?false,?null); ??
19. ????????System.out.println("?[*]?Waiting?for?messages.?To?exit?press?CTRL+C"); ??
20. ??
21. ????????QueueingConsumer?consumer?=?new?QueueingConsumer(channel); ??
22. ????????channel.basicConsume(QUEUE_NAME,?true,?consumer); ??
23. ??
24. ????????while?(true)?{ ??
25. ????????????QueueingConsumer.Delivery?delivery?=?consumer.nextDelivery(); ??
26. ????????????String?message?=?new?String(delivery.getBody()); ??
27. ????????????System.out.println("?[x]?Received?'"?+?message?+?"'"); ??
28. ????????} ??
29. ????} ??
30. }??

package com.abin.test;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

public class Reqv {
	private final static String QUEUE_NAME = "hello";

	public static void main(String[] argv) 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(" [*] Waiting for messages. To exit press CTRL+C");

		QueueingConsumer consumer = new QueueingConsumer(channel);
		channel.basicConsume(QUEUE_NAME, true, consumer);

		while (true) {
			QueueingConsumer.Delivery delivery = consumer.nextDelivery();
			String message = new String(delivery.getBody());
			System.out.println(" [x] Received '" + message + "'");
		}
	}
}

?运行程序得到的结果如下：

1. [*]?Waiting?for?messages.?To?exit?press?CTRL+C ??
2. [x]?Received?'Hello?World!'??

[*] Waiting for messages. To exit press CTRL+C
[x] Received 'Hello World!'

?如果消费者出现“[x] Received 'Hello World!'”说明已接收到此消息信息。