Java编写简易rabbitmq生产者与消费者

一、前言

开发时经常与其它系统用rabbitmq对接，当需要自测时，还是自己写rabbitmq生产者、消费者自测方便些。

下面总结下不用框架、使用java编写简易rabbitmq的方法。

二、代码

1.导入jar包

(1)如果是maven，那就用

		    com.rabbitmq
		    amqp-client
		    3.3.4
		
1
2
3
4
5

(2)如果没用maven，那就找一个amqp-client-3.3.4.jar文件，然后引入项目

2.生产者代码

package RABBIT_MQ;

import com.rabbitmq.client.*;

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


public class RabbitMQSender {

    //交换机
    private final static String exchange_name = "x.mySys.test";
    //路由键
    private final static String route_key = "topmess";

    public static void main(String[] args) throws IOException, TimeoutException {
        //定义一个链接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //设置服务地址
        connectionFactory .setHost("10.123.123.123");
        //设定端口
        connectionFactory .setPort(5673);
        //设定用户名
        connectionFactory .setUsername("myuser");
        //设定密码
        connectionFactory .setPassword("mypass");
        //设定虚拟机
        connectionFactory .setVirtualHost("my_host");

        //从mq工具类获取连接信息
        Connection connection = connectionFactory.newConnection();
        //获取一个通道
        Channel channel = connection.createChannel();
        String message = "Hello World!";
        channel.basicPublish(exchange_name, route_key, null, message.getBytes());
        System.out.println("运行完毕，已发送消息【" + message + "】");

        channel.close();
        connection.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
42
43

说明：
(1)其中的配置，按照自己的rabbitmq配置更改
(2)生产者配置了交换机名与路由键

3.消费者代码

package RABBIT_MQ;

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

import com.rabbitmq.client.AMQP;
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;


public class RabbitMQReceiver {

    private static String queue_name="q.mySys.test";

    public static void main(String[] args) throws IOException, TimeoutException {
        //定义一个链接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //设置服务地址
        connectionFactory .setHost("10.123.123.123");
        //设定端口
        connectionFactory .setPort(5673);
        //设定用户名
        connectionFactory .setUsername("myuser");
        //设定密码
        connectionFactory .setPassword("mypass");
        //设定虚拟机
        connectionFactory .setVirtualHost("my_host");

        //从mq工具类获取连接信息
        Connection connection = connectionFactory.newConnection();
        //获取一个通道
        Channel channel = connection.createChannel();
        //监听该队列，false代表手动确认
        channel.basicConsume(queue_name,false,new DefaultConsumer(channel){
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties basicProperties, byte[] body) throws IOException{
                System.out.println("接收到的消息："+ new String(body,"UTF-8"));
            }
        });
        System.out.println("运行完毕，等待mq接收消息");
    }

}

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
42
43
44
45
46

说明：
(1)其中的配置，按照自己的rabbitmq配置更改
(2)消费者配置了监听队列
(3)消费者启动后一直开启，除非手动关闭项目运行