消息中间件系列教材（十）- RabbitMQ - fanout 模式代码

步骤1:先运行，看到效果，再学习
步骤2:模仿和排错
步骤3:pom.xml
步骤4:RabbitMQUtil
步骤5:TestProducer
步骤6:TestDriectCustomer

步骤 1 : 先运行，看到效果，再学习

老规矩，先下载下载区(点击进入)的可运行项目，配置运行起来，确认可用之后，再学习做了哪些步骤以达到这样的效果。
先运行两次 TestCustomer，启动两个消费者。
然后运行一次 TestProducer，启动生产者，生产100条信息。
此时就可以看到如图所示两个消费者都能收到 这100条信息。

步骤 2 : 模仿和排错

在确保可运行项目能够正确无误地运行之后，再严格照着教程的步骤，对代码模仿一遍。
模仿过程难免代码有出入，导致无法得到期望的运行结果，此时此刻通过比较正确答案 ( 可运行项目 ) 和自己的代码，来定位问题所在。
采用这种方式，学习有效果，排错有效率，可以较为明显地提升学习速度，跨过学习路上的各个槛。

推荐使用diffmerge软件，进行文件夹比较。把你自己做的项目文件夹，和我的可运行项目文件夹进行比较。
这个软件很牛逼的，可以知道文件夹里哪两个文件不对，并且很明显地标记出来
这里提供了绿色安装和使用教程：diffmerge 下载和使用教程

步骤 3 : pom.xml

提供 rabbitmq和hutool的jar

<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

<modelVersion>4.0.0</modelVersion>

<groupId>cn.how2j</groupId>

<artifactId>rabbitmq</artifactId>

<version>0.0.1-SNAPSHOT</version>

<name>rabbitmq</name>

<description>rabbitmq</description>

<dependencies>

<dependency>

<groupId>com.rabbitmq</groupId>

<artifactId>amqp-client</artifactId>

<version>3.6.5</version>

</dependency>

<dependency>

<groupId>cn.hutool</groupId>

<artifactId>hutool-all</artifactId>

<version>4.3.1</version>

</dependency>

</dependencies>

</project>

步骤 4 : RabbitMQUtil

判断服务器是否启动

package cn.how2j;

import javax.swing.JOptionPane;

import cn.hutool.core.util.NetUtil;

public class RabbitMQUtil {

public static void main(String[] args) {

checkServer();

}

public static void checkServer() {

if(NetUtil.isUsableLocalPort(15672)) {

JOptionPane.showMessageDialog(null, "RabbitMQ 服务器未启动 ");

System.exit(1);

}

步骤 5 : TestProducer

发送消息

package cn.how2j;

import java.io.IOException;

import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

/**

* 消息生成者

*/

public class TestProducer {

public final static String EXCHANGE_NAME="fanout_exchange";

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

RabbitMQUtil.checkServer();

//创建连接工厂

ConnectionFactory factory = new ConnectionFactory();

//设置RabbitMQ相关信息

factory.setHost("localhost");

//创建一个新的连接

Connection connection = factory.newConnection();

//创建一个通道

Channel channel = connection.createChannel();

channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

for (int i = 0; i < 100; i++) {

String message = "direct 消息 " +i;

//发送消息到队列中

channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));

System.out.println("发送消息： " + message);

}

//关闭通道和连接

channel.close();

connection.close();

}

步骤 6 : TestDriectCustomer

接受消息

package cn.how2j;

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.Consumer;

import com.rabbitmq.client.DefaultConsumer;

import com.rabbitmq.client.Envelope;

import cn.hutool.core.util.RandomUtil;

public class TestCustomer {

public final static String EXCHANGE_NAME="fanout_exchange";

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

//为当前消费者取随机名

String name = "consumer-"+ RandomUtil.randomString(5);

//判断服务器是否启动

RabbitMQUtil.checkServer();

// 创建连接工厂

ConnectionFactory factory = new ConnectionFactory();

//设置RabbitMQ地址

factory.setHost("localhost");

//创建一个新的连接

Connection connection = factory.newConnection();

//创建一个通道

Channel channel = connection.createChannel();

//交换机声明（参数为：交换机名称；交换机类型）

channel.exchangeDeclare(EXCHANGE_NAME,"fanout");

//获取一个临时队列

String queueName = channel.queueDeclare().getQueue();

//队列与交换机绑定（参数为：队列名称；交换机名称；routingKey忽略）

channel.queueBind(queueName,EXCHANGE_NAME,"");

System.out.println(name +" 等待接受消息");

//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(name + " 接收到消息 '" + message + "'");

}

};

//自动回复队列应答 -- RabbitMQ中的消息确认机制

channel.basicConsume(queueName, true, consumer);

}

更多内容，点击了解： https://how2j.cn/k/message/message-rabbitmq-fanout/2034.html