RabbitMQ的简单实现

╰+哭是因爲堅強的太久メ 2024-04-18 20:07 46阅读 0赞

### 1.简单实现，不使用交换机 ###

#### 引入依赖 ####

<dependency>
                <groupId>com.rabbitmq</groupId>
                <artifactId>amqp-client</artifactId>
                <version>4.8.0</version>
            </dependency>

#### provider ####

import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class Producer {
        public static void main(String[] args) throws IOException, TimeoutException {
    //        创建连接工厂实例
            ConnectionFactory connectionFactory=new ConnectionFactory();
    //        为连接工厂绑定地址、端口、用户名、密码、虚拟主机
            connectionFactory.setHost("192.168.37.134");
            connectionFactory.setPort(5672);
            connectionFactory.setUsername("admin");
            connectionFactory.setPassword("admin");
            connectionFactory.setVirtualHost("/");
    //        取得连接实例
            Connection connection = connectionFactory.newConnection();
    //        取得频道
            Channel channel=connection.createChannel();
    //        频道声明队列信息
            channel.queueDeclare("simple_queue",true,false,false,null);
            /*参数说明
            * String var1,          队列名称
            * boolean var2,         是否将信息持久化，一般为true
            * boolean var3,         是否独占队列，一般为false
            * boolean var4,         信息被消费后是否自动删除，一般为false
            * Map<String, Object> var5      附加参数
            * */
    
            String message="hello luking!";
    //        发送信息
            channel.basicPublish("","simple_queue",null,message.getBytes());
            /*参数说明
            * String var1,      交换机名称，不使用交换机为空
            * String var2,      RoutingKey名称（交换机按此进行派送），或队列名称
            * BasicProperties var5,         附加数据
            * byte[] var6       是要发送的数据
            * */
    //        关闭资源
            channel.close();
            connection.close();
    
        }
    }

#### consumer ####

import com.rabbitmq.client.*;
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class Consumer {
        public static void main(String[] args) throws IOException, TimeoutException {
    //        创建连接工厂实例
            ConnectionFactory connectionFactory=new ConnectionFactory();
    //        为连接工厂绑定地址、端口、用户名、密码、虚拟主机
            connectionFactory.setHost("192.168.37.134");
            connectionFactory.setPort(5672);
            connectionFactory.setUsername("admin");
            connectionFactory.setPassword("admin");
            connectionFactory.setVirtualHost("/");
    //        取得连接实例
            Connection connection = connectionFactory.newConnection();
    //        取得频道
            Channel channel=connection.createChannel();
    //        声明默认消费者
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    
                    System.out.println("ID:" + envelope.getDeliveryTag() + ",   Exchange:" + envelope.getExchange() +
                            ",      RoutingKey:" + envelope.getRoutingKey() + ",      Class:" + envelope.getClass() +
                            body.toString());
                }
            };
    //        接受信息并按方法处理
            channel.basicConsume("simple_queue",true,defaultConsumer);
            /*参数说明
            * String var1,          队列名称
            * boolean var2,         是否自动应答
            * Consumer var3         使用的消费方法实例
            * */
        }
    }

### 2.DIRECT定向交换机 ###

#### provider ####

import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class Producer_DIRECT {
        public static void main(String[] args) throws IOException, TimeoutException {
    //        创建连接工厂实例
            ConnectionFactory connectionFactory=new ConnectionFactory();
    //        为连接工厂绑定地址、端口、用户名、密码、虚拟主机
            connectionFactory.setHost("192.168.37.134");
            connectionFactory.setPort(5672);
            connectionFactory.setUsername("admin");
            connectionFactory.setPassword("admin");
            connectionFactory.setVirtualHost("/");
    //        取得连接实例
            Connection connection = connectionFactory.newConnection();
    //        取得频道
            Channel channel=connection.createChannel();
    //        频道声明队列信息
            channel.queueDeclare("queue_1",true,false,false,null);
            channel.queueDeclare("queue_2",true,false,false,null);
            /*参数说明
            * String var1,          队列名称
            * boolean var2,         是否将信息持久化，一般为true
            * boolean var3,         是否独占队列，一般为false
            * boolean var4,         信息被消费后是否自动删除，一般为false
            * Map<String, Object> var5      附加参数
            * */
    //       声明交换机
            channel.exchangeDeclare("dircet_exchange", BuiltinExchangeType.DIRECT);
            /*声明参数
            *String var1,                   交换机名称
            *BuiltinExchangeType var2       交换机类型：FANOUT（广播），DIRECT（定向），TOPIC（通配符）
            *  */
    //       绑定交换机和通道
            channel.queueBind("queue_1","dircet_exchange","Key.direct.one");
            channel.queueBind("queue_2","dircet_exchange","Key.direct.two");
            
            String message1="hello luking! one";
            String message2="hello luking! two";
    //        发送信息
            channel.basicPublish("dircet_exchange","Key.direct.one",null,message1.getBytes());
            channel.basicPublish("dircet_exchange","Key.direct.two",null,message2.getBytes());
            /*参数说明
            * String var1,      交换机名称，不使用交换机为空
            * String var2,      RoutingKey名称（交换机按此进行派送），或队列名称
            * BasicProperties var5,         附加数据
            * byte[] var6       是要发送的数据
            * */
    //        关闭资源
            channel.close();
            connection.close();
        }
    }

#### consumer ####

import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class Consumer1_DIRECT {
        public static void main(String[] args) throws IOException, TimeoutException {
    //        创建连接工厂实例
            ConnectionFactory connectionFactory=new ConnectionFactory();
    //        为连接工厂绑定地址、端口、用户名、密码、虚拟主机
            connectionFactory.setHost("192.168.37.134");
            connectionFactory.setPort(5672);
            connectionFactory.setUsername("admin");
            connectionFactory.setPassword("admin");
            connectionFactory.setVirtualHost("/");
    //        取得连接实例
            Connection connection = connectionFactory.newConnection();
    //        取得频道
            Channel channel=connection.createChannel();
    //        声明默认消费者，匿名内部类，默认消费者
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    
                    System.out.println("this consume1----ID:" + envelope.getDeliveryTag() + ",   Exchange:" + envelope.getExchange() +
                            ",      RoutingKey:" + envelope.getRoutingKey() + ",      Class:" + envelope.getClass() +
                            new String(body));
                }
            };
    //        接受信息并按方法处理
            channel.basicConsume("queue_1",true,defaultConsumer);
            /*参数说明
            * String var1,          队列名称
            * boolean var2,         是否自动应答
            * Consumer var3         使用的消费方法实例
            * */
        }
    }

### 3.TOPIC通配符交换机 ###

#### provider ####

import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class Producer_TOPIC {
        public static void main(String[] args) throws IOException, TimeoutException {
    //        创建连接工厂实例
            ConnectionFactory connectionFactory=new ConnectionFactory();
    //        为连接工厂绑定地址、端口、用户名、密码、虚拟主机
            connectionFactory.setHost("192.168.37.134");
            connectionFactory.setPort(5672);
            connectionFactory.setUsername("admin");
            connectionFactory.setPassword("admin");
            connectionFactory.setVirtualHost("/");
    //        取得连接实例
            Connection connection = connectionFactory.newConnection();
    //        取得频道
            Channel channel=connection.createChannel();
    //        频道声明队列信息
            channel.queueDeclare("queue_topic_1",true,false,false,null);
            channel.queueDeclare("queue_topic_2",true,false,false,null);
            channel.queueDeclare("queue_topic_3",true,false,false,null);
            /*参数说明
            * String var1,          队列名称
            * boolean var2,         是否将信息持久化，一般为true
            * boolean var3,         是否独占队列，一般为false
            * boolean var4,         信息被消费后是否自动删除，一般为false
            * Map<String, Object> var5      附加参数
            * */
    //      声明交换机
            channel.exchangeDeclare("topic_exchange", BuiltinExchangeType.TOPIC);
            /*声明参数
            *String var1,                   交换机名称
            *BuiltinExchangeType var2       交换机类型：FANOUT（广播），DIRECT（定向），TOPIC（通配符）
            *  */
    //        绑定交换机和通道-----#表示多个单词，*表示一个单词
            channel.queueBind("queue_topic_1","topic_exchange","Key.#");
            channel.queueBind("queue_topic_2","topic_exchange","Key.*.two");
            channel.queueBind("queue_topic_3","topic_exchange","Key.*.one");
    
    
    
            String message1="hello luking! +Key.direct.one";
           /* String message2="hello luking! +Key.direct.two";
            String message3="hello luking! +Key.direct.three";*/
    
    //        发送信息
            channel.basicPublish("topic_exchange","Key.direct.one",null,message1.getBytes());
           // channel.basicPublish("dircet_exchange","Key.direct.two",null,message2.getBytes());
            /*参数说明
            * String var1,      交换机名称，不使用交换机为空
            * String var2,      RoutingKey名称（交换机按此进行派送），或队列名称
            * BasicProperties var5,         附加数据
            * byte[] var6       是要发送的数据
            * */
    //        关闭资源
            channel.close();
            connection.close();
    
        }
    }

#### consumer ####

import com.rabbitmq.client.*;
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class Consumer1_TOPIC {
        public static void main(String[] args) throws IOException, TimeoutException {
    //        创建连接工厂实例
            ConnectionFactory connectionFactory=new ConnectionFactory();
    //        为连接工厂绑定地址、端口、用户名、密码、虚拟主机
            connectionFactory.setHost("192.168.37.134");
            connectionFactory.setPort(5672);
            connectionFactory.setUsername("admin");
            connectionFactory.setPassword("admin");
            connectionFactory.setVirtualHost("/");
    //        取得连接实例
            Connection connection = connectionFactory.newConnection();
    //        取得频道
            Channel channel=connection.createChannel();
    //        声明默认消费者
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    
                    System.out.println("this consume1----ID:" + envelope.getDeliveryTag() + ",   Exchange:" + envelope.getExchange() +
                            ",      RoutingKey:" + envelope.getRoutingKey() + ",      Class:" + envelope.getClass() +
                            new String(body));
                }
            };
    //        接受信息并按方法处理
            channel.basicConsume("queue_topic_1",true,defaultConsumer);
            /*参数说明
            * String var1,          队列名称
            * boolean var2,         是否自动应答
            * Consumer var3         使用的消费方法实例
            * */
        }
    }