八、RabbitMQ 服务异步通讯

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

import org.junit.Test;

import java.io.IOException;

import java.util.concurrent.TimeoutException;

public class PublisherTest {

    @Test

    public void testSendMessage() throws IOException, TimeoutException {

        // 1. 建立连接

*        ConnectionFactory factory = new ConnectionFactory();

        // 1.1. 设置连接参数，分别是：主机名、端口号、vhost、用户名、密码

        factory.setHost("192.168.249.128");

        factory.setPort(5672);

        factory.setVirtualHost("/");

        factory.setUsername("itcast");

        factory.setPassword("123321");

        // 1.2. 建立连接

        Connection connection = factory.newConnection();

        // 2. 创建通道 Channel

        Channel channel = connection.createChannel();

        // 3. 创建队列

        String queueName = "simple.queue";

        channel.queueDeclare(queueName, false, false, false, null);

        // 4. 发送消息

        String message = "hello, rabbitmq!";

        channel.basicPublish("", queueName, null, message.getBytes());

        System.out.println("发送消息成功：【" + message + "】");

        // 5. 关闭通道和连接

        channel.close();

        connection.close();

    }

}

import com.rabbitmq.client.*;

import java.io.IOException;

import java.util.concurrent.TimeoutException;

public class ConsumerTest {

    public static void main(String[] args) throws IOException, TimeoutException {

        // 1. 建立连接

        ConnectionFactory factory = new ConnectionFactory();

        // 1.1. 设置连接参数，分别是：主机名、端口号、vhost、用户名、密码

        factory.setHost("192.168.249.128");

        factory.setPort(5672);

        factory.setVirtualHost("/");

        factory.setUsername("itcast");

        factory.setPassword("123321");

        // 1.2. 建立连接

        Connection connection = factory.newConnection();

        // 2. 创建通道 Channel

        Channel channel = connection.createChannel();

        // 3. 创建队列

        String queueName = "simple.queue";

        channel.queueDeclare(queueName, false, false, false, null);

        // 4. 订阅消息

        channel.basicConsume(queueName, true, new DefaultConsumer(channel){

            @Override

            public void handleDelivery(String consumerTag, Envelope envelope,

                                       AMQP.BasicProperties properties, byte[] body) throws IOException {

                // 5. 处理消息

                String message = new String(body);

                System.out.println("接收到消息：【" + message + "】");

            }

        });

        System.out.println("等待接收消息。。。。");

    }

}

import org.junit.Test;

import org.junit.runner.RunWith;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.boot.test.context.SpringBootTest;

import org.springframework.test.context.junit4.SpringRunner;

/**

 * @author Dkx

 * @version 1.0

 * @2023/10/119:18

 * @function

 * @comment

 */

@SpringBootTest

@RunWith(SpringRunner.class)

public class SpringAmqpTest {

    @Autowired

    private RabbitTemplate rabbitTemplate;

    @Test

    public void test()

    {

        String queueName = "simple.queue";

        String message = "hello,Spring Amqp";

        rabbitTemplate.convertAndSend(queueName, message);

    }

}

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

/**

 * @author Dkx

 * @version 1.0

 * @2023/10/119:34

 * @function

 * @comment

 */

@Component

public class SpringRabbitListener {

    @RabbitListener(queues = "simple.queue")

    public void listenerSimpleQueue(String msg)

    {

        System.out.println("消费者接收到simple.queue的消息为：" + msg);

    }

}

//---------------------- 打印结果 ----------------------

  .   ____          _            __ _ _

 /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \

( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \

 \\/  ___)| |_)| | | | | || (_| |  ) ) ) )

  '  |____| .__|_| |_|_| |_\__, | / / / /

 =========|_|==============|___/=/_/_/_/

 :: Spring Boot ::        (v2.3.9.RELEASE)

10-11 09:37:32:584  INFO 1164 --- [           main] cn.itcast.mq.ConsumerApplication         : Starting ConsumerApplication on Jfier with PID 1164 (E:\微服务\实用篇\资料\day04-MQ\资料\mq-demo\consumer\target\classes started by Administrator in E:\微服务\实用篇\资料\day04-MQ\资料\mq-demo)

10-11 09:37:32:594  INFO 1164 --- [           main] cn.itcast.mq.ConsumerApplication         : No active profile set, falling back to default profiles: default

10-11 09:37:34:476  INFO 1164 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: [192.168.249.128:5672]

10-11 09:37:34:531  INFO 1164 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#640dc4c6:0/SimpleConnection@7a0f244f [delegate=amqp://itcast@192.168.249.128:5672/, localPort= 3380]

10-11 09:37:34:610  INFO 1164 --- [           main] cn.itcast.mq.ConsumerApplication         : Started ConsumerApplication in 2.755 seconds (JVM running for 4.791)

消费者接收到simple.queue的消息为：hello,Spring Amqp

@Test

public void workQueueTest()

{

   String queueName = "simple.queue";

   String message = "hello,Spring Amqp __ ";

   int i = 0;

   for(;i < 50;i++)

   {

      try {

         Thread.sleep(10);

      } catch (InterruptedException e) {

         throw new RuntimeException(e);

      }

      rabbitTemplate.convertAndSend(queueName, message + i);

   }

}

@RabbitListener(queues = "simple.queue")

public void listenerSimpleQueue1(String msg)

{

   System.out.println("消费者1接收到simple.queue的消息为：" + msg);

   try {

      Thread.sleep(20);

   } catch (InterruptedException e) {

      throw new RuntimeException(e);

   }

}

@RabbitListener(queues = "simple.queue")

public void listenerSimpleQueue2(String msg)

{

   System.err.println("消费者2......接收到simple.queue的消息为：" + msg);

   try {

      Thread.sleep(200);

   } catch (InterruptedException e) {

      throw new RuntimeException(e);

   }

}

//-------------------- 打印结果 --------------------

  .   ____          _            __ _ _

 /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \

( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \

 \\/  ___)| |_)| | | | | || (_| |  ) ) ) )

  '  |____| .__|_| |_|_| |_\__, | / / / /

 =========|_|==============|___/=/_/_/_/

 :: Spring Boot ::        (v2.3.9.RELEASE)

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 1---2023-10-11T10:16:47.564155

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 0---2023-10-11T10:16:47.564155

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 3---2023-10-11T10:16:47.602944400

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 5---2023-10-11T10:16:47.634311

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 7---2023-10-11T10:16:47.663563900

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 9---2023-10-11T10:16:47.693902600

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 11---2023-10-11T10:16:47.724277700

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 13---2023-10-11T10:16:47.754720

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 2---2023-10-11T10:16:47.784759800

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 15---2023-10-11T10:16:47.784759800

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 17---2023-10-11T10:16:47.814876500

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 19---2023-10-11T10:16:47.845253300

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 21---2023-10-11T10:16:47.876696200

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 23---2023-10-11T10:16:47.907389600

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 25---2023-10-11T10:16:47.938559800

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 27---2023-10-11T10:16:47.968268500

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 29---2023-10-11T10:16:47.998623300

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 4---2023-10-11T10:16:47.998623300

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 31---2023-10-11T10:16:48.028620600

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 33---2023-10-11T10:16:48.058770700

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 35---2023-10-11T10:16:48.089572100

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 37---2023-10-11T10:16:48.119707200

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 39---2023-10-11T10:16:48.149984400

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 41---2023-10-11T10:16:48.179999600

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 6---2023-10-11T10:16:48.210307400

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 43---2023-10-11T10:16:48.210307400

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 45---2023-10-11T10:16:48.240964800

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 47---2023-10-11T10:16:48.270986500

消费者1接收到simple.queue的消息为：hello,Spring Amqp __ 49---2023-10-11T10:16:48.301165400

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 8---2023-10-11T10:16:48.423433300

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 10---2023-10-11T10:16:48.636403400

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 12---2023-10-11T10:16:48.848982300

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 14---2023-10-11T10:16:49.060187200

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 16---2023-10-11T10:16:49.273197100

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 18---2023-10-11T10:16:49.484165800

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 20---2023-10-11T10:16:49.696880900

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 22---2023-10-11T10:16:49.908464300

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 24---2023-10-11T10:16:50.121477600

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 26---2023-10-11T10:16:50.333586300

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 28---2023-10-11T10:16:50.548236400

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 30---2023-10-11T10:16:50.759585700

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 32---2023-10-11T10:16:50.973508400

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 34---2023-10-11T10:16:51.185203800

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 36---2023-10-11T10:16:51.399855600

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 38---2023-10-11T10:16:51.610433700

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 40---2023-10-11T10:16:51.823352100

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 42---2023-10-11T10:16:52.035381900

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 44---2023-10-11T10:16:52.248226300

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 46---2023-10-11T10:16:52.462320400

消费者2......接收到simple.queue的消息为：hello,Spring Amqp __ 48---2023-10-11T10:16:52.677121500

import org.springframework.amqp.core.Binding;

import org.springframework.amqp.core.BindingBuilder;

import org.springframework.amqp.core.FanoutExchange;

import org.springframework.amqp.core.Queue;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

/**

 * @author Dkx

 * @version 1.0

 * @2023/10/1111:17

 * @function

 * @comment

 */

@Configuration

public class FanoutConfig {

    // itcast.fanout

    @Bean

    public FanoutExchange fanoutExchange()

    {

        return new FanoutExchange("itcast.fanout");

    }

    // fanout.queue1

    @Bean

    public Queue fanoutQueue1()

    {

        return new Queue("fanout.queue1");

    }

    // 绑定队列 1 到交换机

    @Bean

    public Binding fanoutBinding(Queue fanoutQueue1, FanoutExchange fanoutExchange)

    {

        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);

    }

    // fanout.queue2

    @Bean

    public Queue fanoutQueue2()

    {

        return new Queue("fanout.queue2");

    }

    // 绑定队列 2 到交换机

    @Bean

    public Binding fanoutBinding1(Queue fanoutQueue2, FanoutExchange fanoutExchange)

    {

        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);

    }

}

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

/**

 * @author Dkx

 * @version 1.0

 * @2023/10/119:34

 * @function

 * @comment

 */

@Component

public class SpringRabbitListener {

    @RabbitListener(queues = "fanout.queue1")

    public void listenerFanoutQueue1(String msg)

    {

        System.out.println("消费者接收到fanout.queue1的消息为：" + msg);

    }

    @RabbitListener(queues = "fanout.queue2")

    public void listenerFanoutQueue2(String msg)

    {

        System.out.println("消费者接收到fanout.queue2的消息为：" + msg);

    }

}

@Test

public void testFanoutExchange()

{

   // 交换机名称

   String exchangeName = "itcast.fanout";

   // 消息

   String message = "hello , every one";

   // 发送消息，参数分别是：交互机名称，Routingkey (暂时为空)，消息

   rabbitTemplate.convertAndSend(exchangeName, "", message);

}

@RabbitListener(bindings = @QueueBinding(

   // 声明队列

   value = @Queue(name = "direct.queue1"),

   // 声明交换机

   exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),

   // 绑定 key

   key = {"red", "blue"}

))

public void listenDirectQueue1(String msg)

{

   System.out.printf("消费者接收到direct.queue1的消息：[" + msg + "]");

}

@RabbitListener(bindings = @QueueBinding(

   // 声明队列

   value = @Queue(name = "direct.queue1"),

   // 声明交换机

   exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),

   // 绑定 key

   key = {"red", "yellow"}

))

public void listenDirectQueue2(String msg)

{

   System.out.printf("消费者接收到direct.queue2的消息：[" + msg + "]");

}

@Test

public void testDirectExchange()

{

   // 交换机名称

   String exchangeName = "itcast.direct";

   // 消息

   String message = "hello , blue";

   // 发送消息，将消息发送给 key 为 blue 的 Queue

   rabbitTemplate.convertAndSend(exchangeName, "blue", message);

}

//------------------ 打印结果 ------------------

消费者接收到direct.queue1的消息：[hello , blue]

// 发送消息，将消息发送给 key 为 blue 的 Queue

rabbitTemplate.convertAndSend(exchangeName, "yellow", message);

//---------------- 打印结果 ----------------

消费者接收到direct.queue2的消息：[hello , yellow]

// 发送消息，将消息发送给 key 为 blue 的 Queue

rabbitTemplate.convertAndSend(exchangeName, "red", message);

消费者接收到direct.queue2的消息：[hello , red]消费者接收到direct.queue1的消息：[hello , red]

import org.springframework.amqp.core.ExchangeTypes;

import org.springframework.amqp.rabbit.annotation.Exchange;

import org.springframework.amqp.rabbit.annotation.Queue;

import org.springframework.amqp.rabbit.annotation.QueueBinding;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

/**

 * @author Dkx

 * @version 1.0

 * @2023/10/119:34

 * @function

 * @comment

 */

@Component

public class SpringRabbitListener {

    @RabbitListener(bindings = @QueueBinding(

            value = @Queue(name = "topic.queue1"),

            exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),

            key = "china.#"

    ))

    public void listenTopicQueue1(String msg)

    {

        System.out.println("消费者接收到topic.queue1的消息：[" + msg + "]");

    }

    @RabbitListener(bindings = @QueueBinding(

            value = @Queue(name = "topic.queue2"),

            exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),

            key = "#.news"

    ))

    public void listenTopicQueue2(String msg)

    {

        System.out.println("消费者接收到topic.queue2的消息：[" + msg + "]");

    }

}

@Test

public void testSendTopicExchange()

{

   String exchangeName = "itcast.topic";

   String message = "郭明然上市了不要888不要999只要9.9就能带回家";

   rabbitTemplate.convertAndSend(exchangeName, "china.news", message);

}

//-------------------------- 打印结果 --------------------------

消费者接收到topic.queue2的消息：[郭明然上市了不要888不要999只要9.9就能带回家]

消费者接收到topic.queue1的消息：[郭明然上市了不要888不要999只要9.9就能带回家]

import org.springframework.amqp.core.Queue;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

/**

 * @author Dkx

 * @version 1.0

 * @2023/10/1111:17

 * @function

 * @comment

 */

@Configuration

public class FanoutConfig {

    @Bean

    public Queue objectQueue()

    {

        return new Queue("object.queue");

    }

}

@Test

public void testSendObjectQueue()

{

   Map<String, Object> map = new HashMap<>();

   map.put("name", "刘桑");

   map.put("age", 18);

   rabbitTemplate.convertAndSend("object.queue", map);

}

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;

import org.springframework.amqp.support.converter.MessageConverter;

import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;

import org.springframework.context.annotation.Bean;

@SpringBootApplication

public class PublisherApplication {

    public static void main(String[] args) {

        SpringApplication.run(PublisherApplication.class);

    }

    @Bean

    public MessageConverter messageConverter()

    {

        return new Jackson2JsonMessageConverter();

    }

}

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;

import org.springframework.amqp.support.converter.MessageConverter;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

/**

 * @author Dkx

 * @version 1.0

 * @2023/10/1117:31

 * @function

 * @comment

 */

@Configuration

public class RabbitMQConfig {

    @Bean

    public MessageConverter messageConverter()

    {

        return new Jackson2JsonMessageConverter();

    }

}