|
|
|
步骤 2 : 模仿和排错 步骤 3 : pom.xml 步骤 4 : RabbitMQUtil 步骤 5 : TestProducer 步骤 6 : TestCustomer4USA 步骤 7 : TestCustomer4News
老规矩，先下载右上角的可运行项目，配置运行起来，确认可用之后，再学习做了哪些步骤以达到这样的效果。
先运行 TestCustomer4USA 专门用于接受美国专题消息 再运行 TestCustomer4News 专门用于接受新闻专题消息 最后运行 TestProducer ，分别在 四个路由："usa.news", "usa.weather", "europe.news", "europe.weather" 上发布 "美国新闻", "美国天气", "欧洲新闻", "欧洲天气". 于是就能在消费者端看到 不同的主题收到对应的消息了。
在确保可运行项目能够正确无误地运行之后，再严格照着教程的步骤，对代码模仿一遍。
模仿过程难免代码有出入，导致无法得到期望的运行结果，此时此刻通过比较正确答案 ( 可运行项目 ) 和自己的代码，来定位问题所在。 采用这种方式，学习有效果，排错有效率，可以较为明显地提升学习速度，跨过学习路上的各个槛。 推荐使用diffmerge软件，进行文件夹比较。把你自己做的项目文件夹，和我的可运行项目文件夹进行比较。 这个软件很牛逼的，可以知道文件夹里哪两个文件不对，并且很明显地标记出来 这里提供了绿色安装和使用教程：diffmerge 下载和使用教程
提供相关 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>
判断服务器是否启动
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);
}
}
}
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);
}
}
}
分别在 四个路由："usa.news", "usa.weather", "europe.news", "europe.weather" 上发布 "美国新闻", "美国天气", "欧洲新闻", "欧洲天气".
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="topics_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, "topic");
String[] routing_keys = new String[] { "usa.news", "usa.weather",
"europe.news", "europe.weather" };
String[] messages = new String[] { "美国新闻", "美国天气",
"欧洲新闻", "欧洲天气" };
for (int i = 0; i < routing_keys.length; i++) {
String routingKey = routing_keys[i];
String message = messages[i];
channel.basicPublish(EXCHANGE_NAME, routingKey, null, message
.getBytes());
System.out.printf("发送消息到路由：%s, 内容是: %s%n ", routingKey,message);
}
//关闭通道和连接
channel.close();
connection.close();
}
}
专门用于接受 usa.* 消息
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 TestCustomer4USA {
public final static String EXCHANGE_NAME="topics_exchange";
public static void main(String[] args) throws IOException, TimeoutException {
//为当前消费者取名称
String name = "consumer-usa";
//判断服务器是否启动
RabbitMQUtil.checkServer();
// 创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
//设置RabbitMQ地址
factory.setHost("localhost");
//创建一个新的连接
Connection connection = factory.newConnection();
//创建一个通道
Channel channel = connection.createChannel();
//交换机声明（参数为：交换机名称；交换机类型）
channel.exchangeDeclare(EXCHANGE_NAME,"topic");
//获取一个临时队列
String queueName = channel.queueDeclare().getQueue();
//接受 USA 信息
channel.queueBind(queueName, EXCHANGE_NAME, "usa.*");
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);
}
}
专门用于接受 *.news 消息
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 TestCustomer4News {
public final static String EXCHANGE_NAME="topics_exchange";
public static void main(String[] args) throws IOException, TimeoutException {
//为当前消费者取名称
String name = "consumer-news";
//判断服务器是否启动
RabbitMQUtil.checkServer();
// 创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
//设置RabbitMQ地址
factory.setHost("localhost");
//创建一个新的连接
Connection connection = factory.newConnection();
//创建一个通道
Channel channel = connection.createChannel();
//交换机声明（参数为：交换机名称；交换机类型）
channel.exchangeDeclare(EXCHANGE_NAME,"topic");
//获取一个临时队列
String queueName = channel.queueDeclare().getQueue();
//接受 USA 信息
channel.queueBind(queueName, EXCHANGE_NAME, "*.news");
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);
}
}
HOW2J公众号，关注后实时获知最新的教程和优惠活动，谢谢。
问答区域
2020-04-09
kafka消息队列教程啥时出啊
2019-12-24
内容啥时候扩充一下呢
回答已经提交成功，正在审核。 请于 我的回答 处查看回答记录，谢谢
2019-05-17
站长大大 何时出一个kafka消息队列 教程哈
提问之前请登陆
提问已经提交成功，正在审核。 请于 我的提问 处查看提问记录，谢谢
|
|||||||||||
