public class ReceiveLogs02 {
private static final String EXCHANGE_NAME = "logs"; public static void main(String[] argv) throws Exception {
Channel channel = RabbitUtils.getChannel(); channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
/**
* 生成一个临时的队列 队列的名称是随机的
* 当消费者断开和该队列的连接时 队列自动删除
*/
String queueName = channel.queueDeclare().getQueue();
//把该临时队列绑定我们的 exchange 其中 routingkey(也称之为 binding key)为空字符串 channel.queueBind(queueName, EXCHANGE_NAME, "");
System.out.println("等待接收消息,把接收到的消息写到文件 ........... "); DeliverCallback deliverCallback = (consumerTag, delivery) ->
{ String message = new String(delivery.getBody(), "UTF-8"); File file = new File("C:\\work\\rabbitmq_info.txt"); FileUtils.writeStringToFile(file,message,"UTF-8");
System.out.println("数据写入文件成功");
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });
}
}

public class EmitLog {
private static final String EXCHANGE_NAME = "logs"; public static void main(String[] argv) throws Exception {
try (Channel channel = RabbitUtils.getChannel()) {
/**
*  声明一个 exchange
*  1.exchange 的名称
*  2.exchange 的类型
*/
channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); Scanner sc = new Scanner(System.in);
System.out.println("请输入信息");
while (sc.hasNext()) {
String message = sc.nextLine();
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8")); System.out.println("生产者发出消息" + message);
}
}
}
}

package com.atguigu.rabbitmq.five;
import com.atguigu.rabbitmq.utils.RabbitMqutils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ReceiveLogs01 {
    //交换机的名字
    public static final String EXCHANGE_NAME = "logs";
    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMqutils.getChannel();
        //声明一个交换机
        /**
         * 1.交换机的名字
         * 2.类型 fanout 主要练习的就是他
         */
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        /**
         * 声明一个队列  临时的（队列的名称随机的）
         * 连接的时候就在 不连接的时候就不在了
         * 断开队列的时候自动删除
         */
        String queueName = channel.queueDeclare().getQueue();
        /**
         * 绑定交换机与队列
         * 1.队列名称
         * 2.交换机的名称
         * 3.routingkey 空 可以空
         */
        channel.queueBind(queueName, EXCHANGE_NAME, "");
        System.out.println("等待接受消息把接受到的消息打印到屏幕上面....");
        //接受消息
        DeliverCallback deliverCallback = (var1, var2) ->
        {
            System.out.println("ReceiveLogs01控制台打印接口道的消息：-[]" + new String(var2.getBody(), "utf-8"));
        };
        //消费者取消消息时回调的接口
        channel.basicConsume(queueName, true, deliverCallback, var1 -> {
        }, null);
    }
}

package com.atguigu.rabbitmq.five;
import com.atguigu.rabbitmq.utils.RabbitMqutils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ReceiveLogs02 {
    //交换机的名字
    public static final String EXCHANGE_NAME = "logs";
    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMqutils.getChannel();
        //声明一个交换机
        /**
         * 1.交换机的名字
         * 2.类型 fanout 主要练习的就是他
         */
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        /**
         * 声明一个队列  临时的（队列的名称随机的）
         * 连接的时候就在 不连接的时候就不在了
         * 断开队列的时候自动删除
         */
        String queueName = channel.queueDeclare().getQueue();
        /**
         * 绑定交换机与队列
         * 1.队列名称
         * 2.交换机的名称
         * 3.routingkey 空 可以空
         */
        channel.queueBind(queueName, EXCHANGE_NAME, "");
        System.out.println("等待接受消息把接受到的消息打印到屏幕上面....");
        //接受消息
        DeliverCallback deliverCallback = (var1, var2) ->
        {
            System.out.println("ReceiveLogs02控制台打印接口道的消息：-[]" + new String(var2.getBody(), "utf-8"));
        };
        //消费者取消消息时回调的接口
        channel.basicConsume(queueName, true, deliverCallback, var1 -> {
        }, null);
    }
}

package com.atguigu.rabbitmq.five;
import com.atguigu.rabbitmq.utils.RabbitMqutils;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.util.Scanner;
import java.util.concurrent.TimeoutException;
/**
 * 发消息给交换机
 */
public class EmitLog {
    //交换机的名字
    public static final String EXCHANGE_NAME = "logs";
    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMqutils.getChannel();
        //交互机
       // channel.exchangeDeclare(EXCHANGE_NAME,"fauout");
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()){
            String message = scanner.next();
            channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes("utf-8"));
            System.out.println("生产者发出的消息是什么"+message);
        }
    }
}