1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
<dependencies>
	<dependency>
		<groupId>com.rabbitmq</groupId>
		<artifactId>amqp-client</artifactId>
		<version>4.0.2</version>
	</dependency>
	<dependency>
		<groupId>org.slf4j</groupId>
		<artifactId>slf4j-api</artifactId>
		<version>1.7.10</version>
	</dependency>
	<dependency>
		<groupId>org.slf4j</groupId>
		<artifactId>slf4j-log4j12</artifactId>
		<version>1.7.5</version>
	</dependency>
	<dependency>
		<groupId>log4j</groupId>
		<artifactId>log4j</artifactId>
		<version>1.2.17</version>
	</dependency>
	<dependency>
		<groupId>junit</groupId>
		<artifactId>junit</artifactId>
		<version>4.11</version>
	</dependency>
</dependencies>

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import java.io.IOException;
import java.util.concurrent.TimeoutException;

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

public class MqConnectUtil {
	
	/**
	 * 获取mq的连接
	 * @return
	 * @throws TimeoutException 
	 * @throws IOException 
	 */
	public static Connection getMqConnection() throws IOException, TimeoutException{
		//1、定义一个连接工厂
		ConnectionFactory factory = new ConnectionFactory();
		//2、设置连接地址等信息
		factory.setHost("127.0.0.1");
		//amqp协议端口
		factory.setPort(5672);
		//设置具体的vhost(想象成数据库)
		factory.setVirtualHost("/xiangjiao");
		//设置用户名密码信息
		factory.setUsername("xiangjiao");
		factory.setPassword("bunana");
		
		return factory.newConnection();
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import cn.linkpower.util.MqConnectUtil;

/**
 * 简单消息队列 生产者
 * @author 76519
 *
 */
public class Send {
	private static String simpleQueueName = "test_simple_queue";
	public static void main(String[] args) throws Exception{
		//创建一个连接
		Connection conn = MqConnectUtil.getMqConnection();
		//构建一条通道 --- 从连接中获取一个通道
		Channel createChannel = conn.createChannel();
		//申明一个指定的队列
		createChannel.queueDeclare(simpleQueueName,
				false, false, false, null);
		
		//发送数据信息
		String msg = "hello xiangjiao bunana";
		//发送数据
		createChannel.basicPublish("", 
				simpleQueueName, null, msg.getBytes());
		
		System.out.println("---send msg = "+msg);
		
		//关闭流和连接
		createChannel.close();
		conn.close();
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;
import com.rabbitmq.client.ShutdownSignalException;

import cn.linkpower.util.MqConnectUtil;

/**
 * 使用程序代码获取send端发送的消息(从mq中自动取得)
 * @author 76519
 *
 */
public class GetMqMsg_old {
	private static String simpleQueueName = "test_simple_queue";
	public static void main(String[] args) throws IOException, TimeoutException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
		//获取连接
		Connection mqConnection = MqConnectUtil.getMqConnection();
		//构建通道
		Channel createChannel = mqConnection.createChannel();
		
		//定义一个队列的消费者
		QueueingConsumer queueingConsumer = new QueueingConsumer(createChannel);
		
		//监听
		createChannel.basicConsume(simpleQueueName, true,queueingConsumer);
		
		//获取数据
		while(true){
			
			Delivery nextDelivery = queueingConsumer.nextDelivery();
			String msg = new String(nextDelivery.getBody());
			System.out.println("get old msg == "+msg);
		}
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
import java.io.IOException;
import java.util.concurrent.TimeoutException;

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

import cn.linkpower.util.MqConnectUtil;

public class GetMqMsg_new {
	private static String simpleQueueName = "test_simple_queue";

	public static void main(String[] args) throws IOException, TimeoutException {
		// 获取连接
		Connection mqConnection = MqConnectUtil.getMqConnection();
		// 构建通道
		Channel createChannel = mqConnection.createChannel();

		DefaultConsumer consumer = new DefaultConsumer(createChannel) {
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
					throws IOException {
				String message = new String(body, "UTF-8");
				System.out.println(" get msg new = " + message );
			}
		};
		//监听队列
		createChannel.basicConsume(simpleQueueName, true, consumer);
	}
}

1
2
3
createChannel.queueDeclare(simpleQueueName, 
				false, false, false, null);

1
2
3
createChannel.queueDeclare(simpleQueueName, 
				false, false, false, null);

1
2
3
4
5
6
Queue.DeclareOk queueDeclare(String queue, 
boolean durable, 
boolean exclusive, 
boolean autoDelete,
Map<String, Object> arguments) throws IOException;

1
2
channel.basicConsume(queue_name, false,consumer);