八.spring+rabbitmq

深藏阁楼爱情的钟 2022-06-12 00:27 133阅读 0赞

**一.spring+rabbitmq使用main方法集成**  
1.pom.xml

<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>com.tiglle</groupId>
      <artifactId>spring-rabbitmq-main</artifactId>
      <version>0.0.1-SNAPSHOT</version>
    
      <dependencies>
      
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit</artifactId>
            <version>1.7.1.RELEASE</version>
        </dependency>
        
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
            <version>1.2.1</version>
        </dependency>
      </dependencies>
    
    
    </project>

2.Producer.java:

package com.rabbit.producer.main;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.core.TopicExchange;
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.core.RabbitAdmin;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    
    
    public class Producer {
    
        private static Logger logger = LoggerFactory.getLogger(Producer.class);
    
        public static void main(String[] args) {
            //获取一个连接工厂，用户默认是guest/guest（只能使用部署在本机的RabbitMQ）
            //是Spring实现的对com.rabbitmq.client.Connection的包装
            ConnectionFactory cf = new CachingConnectionFactory("localhost");
    
            //对AMQP 0-9-1的实现
            RabbitAdmin admin = new RabbitAdmin(cf);
            //声明一个队列
            Queue queue = new Queue("myQueue");
            admin.declareQueue(queue);
            //声明一个exchange类型为topic
            TopicExchange exchange = new TopicExchange("myExchange");
            admin.declareExchange(exchange);
            //绑定队列到exchange，并指定routingKey为foo.*
            admin.declareBinding(BindingBuilder.bind(queue).to(exchange).with("foo.*"));
    
            //发送模版，设置上连接工厂
            RabbitTemplate template = new RabbitTemplate(cf);
            //发送消息
            /** * 1.String exchange：exchange的名称 * 2.String routingKey:routingKey的名称 * 3.Object message:要像exchange发送的消息 */
            template.convertAndSend("myExchange", "foo.bar", "Hello Tiglle");
    
            logger.info("Produce发送消息到"+exchange.getName()+"的exchange上,"
                    + "queueName="+queue.getName()+",routingKey=foo.*");
        } 
    
    }

3.Consumer.java:

package com.rabbit.comsumer.main;
    
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
    import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
    
    public class Comsumer {
    
        public static void main(String[] args) {
            //获取一个连接工厂，用户默认是guest/guest（只能使用部署在本机的RabbitMQ）
            //是Spring实现的对com.rabbitmq.client.Connection的包装
            ConnectionFactory cf = new CachingConnectionFactory("localhost");
            //监听容器
            SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(cf);
            //监听者对象
            Object listener = new Object() {
                @SuppressWarnings("unused")
                public void handleMessage(String foo) {
                    System.out.println(foo);
                }
            };
           //通过这个适配器代理listener
            MessageListenerAdapter adapter = new MessageListenerAdapter(listener);
            //把适配器（listener）设置给Container
            container.setMessageListener(adapter);
            //设置该容器监听的队列名，可以传多个，public void setQueueNames(String... queueName) 
             container.setQueueNames("myQueue");
            //开始监听
             container.start();
        }
    
    
    }

启动Procuder和Consumer可以成功发送接收消息

**二.通过配置文件配置**  
1.pom.xml

<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>com.tiglle</groupId>
      <artifactId>rabbitmq-spring</artifactId>
      <version>0.0.1-SNAPSHOT</version>
    
      <dependencies>
        <dependency>
            
            <groupId>org.springframework</groupId>
            <artifactId>spring-context</artifactId>
            <version>4.3.7.RELEASE</version>
        </dependency>
    
        
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit</artifactId>
            <version>1.7.1.RELEASE</version>
        </dependency>
    
        
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
            <version>1.2.1</version>
        </dependency>
      </dependencies>
    
    </project>

2.spring+rabbitmq配置文件:applicationContext-rabbit.xml

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context" xmlns:rabbit="http://www.springframework.org/schema/rabbit" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.2.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.6.xsd ">
    
        
        
        <context:annotation-config/>
        
        <context:component-scan base-package="com.rabbit"/>
    
    
    
        
        
        <rabbit:connection-factory id="connectionFactory" host="127.0.0.1" username="guest" password="guest" port="5672" />
    
        
        <rabbit:admin connection-factory="connectionFactory" />
    
        
        <rabbit:queue name="myQueue" />
    
        
        <rabbit:topic-exchange name="myExchange">
            <rabbit:bindings>
                <rabbit:binding queue="myQueue" pattern="core.*" />
                
            </rabbit:bindings>
        </rabbit:topic-exchange>
    
         
         <rabbit:template id="amqpTemplate" connection-factory="connectionFactory" exchange="myExchange"/>
    
        
        <rabbit:listener-container connection-factory="connectionFactory">
            
            <rabbit:listener ref="consumer" method="consumerMessage" queue-names="myQueue"/>
            
        </rabbit:listener-container>
    
    </beans>

3.消费者，通过注解注入spring容器中的:Consumer.java

package com.rabbit.consumer;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.stereotype.Component;
    //注入spring容器
    @Component
    public class Consumer { //配置文件raf的类
    
        Logger logger = LoggerFactory.getLogger(Consumer.class);
        //配置文件指定的消息处理的方法
        public void consumerMessage(String message){
            logger.info("接收的消息为:"+message);
        }
    
    }

4.测试启动spring并发送消息的Mian方法：ProducerMain.java

package com.rabbit.main;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.context.support.AbstractApplicationContext;
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    
    import com.rabbit.consumer.Consumer;
    
    public class ProducerMain {
    
        static Logger logger = LoggerFactory.getLogger(ProducerMain.class);
    
        //发送者
        public static void main(String[] args) throws InterruptedException {
            //启动spring容器,启动后消费者就会一直监听
            AbstractApplicationContext beans = new ClassPathXmlApplicationContext("applicationContext.xml");
    
            //假装是Autowrited的
            //@Autowrited
            RabbitTemplate rabbitTemplate = beans.getBean(RabbitTemplate.class);
            //设置routingKey
            rabbitTemplate.setRoutingKey("core.info");
            //发送,exchange，routingKey都在配置文件中配置好了
            rabbitTemplate.convertAndSend("hellow tiglle");
            logger.info("发送的消息为:hellow tiglle");
    
            //关闭掉spring容器
            Thread.sleep(1000);
            beans.destroy();
        }
    
    }