POM.XML

<dependency>

<groupId>com.rabbitmq</groupId>

<artifactId>amqp-client</artifactId>

<version>4.2.1</version>

</dependency>

RabbitMQProduct.java

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

/**
 * 生產者   向RabbitMQ傳送訊息
 */
public class RabbitMQProduct {

	private static final String QUEUE_NAME="demo.queue";
	private static final String EXCHANGE_NAME="demo.exchange";
	
	public static void main(String[] args) throws Exception {
		ConnectionFactory factory = new ConnectionFactory(); //連線工廠  
		//配置連線引數  資訊
		factory.setUsername("rabbitstudy");
		factory.setPassword("123456");
		factory.setHost("192.168.0.1");// 自己的RabbitMQ的ip地址
		factory.setPort(5672);
		Connection connection = factory.newConnection(); //建立連線
		Channel channel = connection.createChannel(); //建立通道 在通道上傳遞訊息
		channel.exchangeDeclare(EXCHANGE_NAME, "direct"); //建立交換器
		channel.queueDeclare(QUEUE_NAME, false, false, false, null); //建立佇列
		channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "com.cdsn.test"); //通過路由鍵繫結交換器與佇列
		//傳送訊息給RabbitMQ
		channel.basicPublish(EXCHANGE_NAME, "com.cdsn.test", 
				MessageProperties.PERSISTENT_TEXT_PLAIN, 
				"Hello World!".getBytes());
		//關閉資源
		channel.close();
		connection.close();
	}
}
 

RabbitMQConsumer.java

​

import java.io.IOException;
import java.util.concurrent.TimeUnit;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * 消費者   處理RabbitMQ中的訊息
 */
public class RabbitMQConsumer {

	private static final String QUEUE_NAME="demo.queue";
	public static void main(String[] args) throws Exception {
		ConnectionFactory factory = new ConnectionFactory(); //連線工廠  
		//配置連線引數  資訊
		factory.setUsername("rabbitstudy");
		factory.setPassword("123456");
		factory.setHost("192.168.0.1");// 自己RabbitMQ的連線地址
		factory.setPort(5672);
		Connection connection = factory.newConnection(); //建立連線
		final Channel channel = connection.createChannel(); //建立通道 在通道上傳遞訊息
		//告訴RabbitMQ  我可以接收訊息了
		channel.basicConsume(QUEUE_NAME, new DefaultConsumer(channel){
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
					throws IOException {
				System.out.println("消費者接收到："+new String(body));
				//告訴伺服器， 我收到訊息了
				channel.basicAck(envelope.getDeliveryTag(), false); 
			}
		});
		TimeUnit.SECONDS.sleep(1);
		//關閉
		channel.close();
		connection.close();
	}
}