comp-example/pom.xml

@@ -54,6 +54,12 @@
		</dependency>
		<dependency>
			<groupId>com.rabbitmq</groupId>
			<artifactId>amqp-client</artifactId>
			<version>3.6.5</version>
		</dependency>
		<dependency>
			<groupId>com.ai.ipu</groupId>
			<artifactId>ipu-zk</artifactId>
			<version>${ipu}</version>

comp-example/src/test/java/com/ai/ipu/example/rabbitmq/RabbitMQCustomerExample.java

@@ -0,0 +1,45 @@
package com.ai.ipu.example.rabbitmq;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
 * @author liutong3
 * @team IPU
 * @date 2019/5/28 11:00
 * @desc rabbitmq消费者
 */
public class RabbitMQCustomerExample {
    private final static String QUEUE_NAME = "rabbitMQ.test";
    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置RabbitMQ地址
        factory.setHost("47.105.160.21");
        factory.setUsername("admin");
        factory.setPassword("admin");
        //创建一个新的连接
        Connection connection = factory.newConnection();
        //创建一个通道
        Channel channel = connection.createChannel();
        //声明要关注的队列
        channel.queueDeclare(QUEUE_NAME, false, false, true, null);
        //DefaultConsumer类实现了Consumer接口，通过传入一个频道，
        // 告诉服务器我们需要那个频道的消息，如果频道中有消息，就会执行回调函数handleDelivery
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("返回消息信息：");
                System.out.println(message);
            }
        };
        //自动回复队列应答 -- RabbitMQ中的消息确认机制
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

comp-example/src/test/java/com/ai/ipu/example/rabbitmq/RabbitMQProducerExample.java

@@ -0,0 +1,50 @@
package com.ai.ipu.example.rabbitmq;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.io.IOException;
import java.util.Scanner;
import java.util.concurrent.TimeoutException;
/**
 * @author liutong3
 * @team IPU
 * @date 2019/5/28 11:02
 * @desc tabbitmq生产者
 */
public class RabbitMQProducerExample {
    public final static String QUEUE_NAME="rabbitMQ.test";
    public static void main(String[] args) throws IOException, TimeoutException {
        //创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置RabbitMQ相关信息
        factory.setHost("47.105.160.21");
        factory.setUsername("admin");
        factory.setPassword("admin");
//        factory.setPort();
        //创建一个新的连接
        Connection connection = factory.newConnection();
        //创建一个通道
        Channel channel = connection.createChannel();
        //  声明一个队列        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        try {
            Scanner sc = new Scanner(System.in);
            int i = 0;
            while (true) {
                i++;
                System.out.println("请输入第" + i + "条信息的内容：");
                String message = sc.nextLine();
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
                System.out.println("Sent:" + message);
            }
        } finally {
            //关闭通道和连接
            channel.close();
            connection.close();
        }
    }
}