生产端和消费端的代码实现，创建一个辅助类，用以配置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