生产端和消费端的代码实现，创建一个辅助类，用以配置RabbitMQ和发送消息，代码如下：
/**
* RabbitMQ 消息推送
* RabbitMQ 后台管理器地址：localhost：15672 可以查看消息的情况
*/
public class RabbitMQUtils2 {
private static String userName = "test";//用户名
private static String passWord = "test123";//用户密码
private static String hostName = "";//服务器IP
private static int portNum = 5672;//端口 默认为5672
private static String exchangeName = "test_exchange";//交换机名称
private static ConnectionFactory factory = new ConnectionFactory();//连接工厂对象 创建新连接
private static final String routingKey = "test_rout";//路由名称
private static String queueName = "test_queue";//队列名称 接收端和发送端分别从指定的队列去去消息
/**
* Rabbit配置
*/
public static void setupConnectionFactory() {
factory.setUsername(userName);//用户名
factory.setPassword(passWord);//密码
factory.setHost(hostName);//服务器ip
factory.setPort(portNum);//端口
}
/**
* 发消息
*/
public static void basicPublish() {
new Thread(){
@Override
public void run() {
super.run();
try {
//连接 通过连接工厂创建一个新连接 与接收端不是同一个
Connection connectionP = factory.newConnection();
//通道 通过新连接创建一个新通道 与接收端也不是同一个
Channel channelP = connectionP.createChannel();
//声明了一个交换和一个服务器命名的队列，然后将它们绑定在一起
channelP.exchangeDeclare(exchangeName,"direct",true);//声明exchange
channelP.queueBind(queueName,exchangeName,routingKey);//将exchange和queue绑定
//消息发布
byte[] msg = "hello world!".getBytes();
//rountingKey
channelP.basicPublish(exchangeName, routingKey, null, msg);
//关闭通道
channelP.close();
//关闭连接
connectionP.close();
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
}
}.start();
}
/**
* 收消息
*/
public static void basicConsume() {
new Thread(){
@Override
public void run() {
super.run();
try {
//连接 通过连接工厂创建一个新连接 与发送端不是同一个
Connection connectionC = factory.newConnection();
//通道 通过新连接创建一个新通道 与发送端也不是同一个
final Channel channelC = connectionC.createChannel();
//声明了一个交换和一个服务器命名的队列，然后将它们绑定在一起。
channelC.exchangeDeclare(exchangeName, "direct", true);
channelC.queueBind(queueName, exchangeName, routingKey);
//实现Consumer的最简单方法是将便捷类DefaultConsumer子类化。可以在basicConsume 调用上传递此子类的对象以设置订阅：
channelC.basicConsume(queueName, false, "administrator", new DefaultConsumer(channelC) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
//接收到的消息
String msg = new String(body, "UTF-8");
//交付标记
long deliveryTag = envelope.getDeliveryTag();
channelC.basicAck(deliveryTag, false);
Log.e("TAG","接收的消息:"+msg);
}
});
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
}
}.start();
}
}
代码中已经给了详细的注释，大致流程就是创建连接>创建信道>声明交换机和队列>将交换机、队列以及他们匹配的路由绑定在一起，最后就是消息的发送和接收。
activity布局只有一个按钮，点击按钮模拟发送消息，接收的消息早控制台打印出来，下面是activity的代码：
public class MainActivity extends AppCompatActivity {
private Button button_p;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
button_p = findViewById(R.id.btn_p);
button_p.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
RabbitMQUtils2.basicPublish();//发送消息
}
});
// RabbitMQUtils2.basicConsume();//接收消息
}
}
程序运行起来，我们可以先把接收消息注释掉，先不消费，发送事件后，我们可以去后台管理看看消息队列的变化：
消息队列
这就表示生产者发送了一条消息，消费者没有消费这条消息，那么消息就会一直存在与队列中，消息的变化过程在概况总览中有两个折线图很形象的表示了出来：
概况总览
图中可以看到，如果消息没有得到消费就一直会存在队列中。接下来我们将消费者注释放开，让其能够消费消息，再来看看折线的变化：
消息的产生与消费
以上是后台管理看到的变化，接下来我们再去控制台（消费者）看看，我们是否有打印出发送的消息：
image.png
“hello world！”这个是生产者发送的消息，至此RabbitMQ整个使用过程已经讲述完毕。整个过程是一步一个坑踩过来的，在下一篇文章RabbitMQ使用过程中遇到的问题
里主要讲述在过程中遇到的问题以及解决的方法。
