spring boot 使用rabbitmq

怼烎@ 2022-10-09 03:07 154阅读 0赞

### spring boot 使用 rabbitmq ###

*   *   *   *  配置和工作流程
             *  使用注解
             *  功能说明
             *  代码实现

#### 配置和工作流程 ####

server.port=8081
    logging.path=D:/
    logging.level.boot.spring=DEBUG
    spring.application.name=spring-boot-rabbitmq
    spring.rabbitmq.host=localhost
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest

创建项目中使用的 队列，交换机，路由 并设置 路由的消息传播方式

rabbitmq 经常使用的有3种

直连模式

@Bean
     	public DirectExchange directExchange(){ 
    		DirectExchange directExchange=new DirectExchange("direct");
     		return directExchange;
     	}

广播模式

@Bean
     	public FanoutExchange fanoutExchange(){ 
     		FanoutExchange fanoutExchange=new FanoutExchange("fanout");
     		return fanoutExchange;
     	}

主题模式

@Bean
     	public TopicExchange topicExchange(){ 
    		TopicExchange topicExchange=new TopicExchange("mytopic");
     		return topicExchange;
     	}

创建队列

@Bean
        public Queue topicQueue2() { 
           Queue queue=new Queue("topicqueue2");
           return queue;
        }

绑定队列和路由

@Bean
     	public Binding bindingtopic1(){ 
     		Binding binding=BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("*.orange.*");//binding key
     		return binding;
     	}

完整的示例

package boot.spring.config;
    
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.core.TopicExchange;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    //topic交换机模型，需要一个topic交换机，两个队列和三个binding
    @Configuration
    public class TopicExchangeConfig { 
    	@Bean
     	public TopicExchange topicExchange(){ 
    		TopicExchange topicExchange=new TopicExchange("mytopic");
     		return topicExchange;
     	}
    	
    	@Bean
        public Queue topicQueue1() { 
           Queue queue=new Queue("topicqueue1");
           return queue;
        }
     	
     	@Bean
        public Queue topicQueue2() { 
           Queue queue=new Queue("topicqueue2");
           return queue;
        }
    	
     	//3个binding将交换机和相应队列连起来
     	@Bean
     	public Binding bindingtopic1(){ 
     		Binding binding=BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("*.orange.*");//binding key
     		return binding;
     	}
     	
     	@Bean
     	public Binding bindingtopic2(){ 
     		Binding binding=BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("*.*.rabbit");
     		return binding;
     	}
     	
     	@Bean
     	public Binding bindingtopic3(){ 
     		Binding binding=BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("lazy.#");//#表示0个或若干个关键字，*表示一个关键字
     		return binding;
     	}
    }

创建完 队列，路由，交换机 并相互绑定以后

就可以使用 rabbitTeplate 的 convertAndSend 方法相队列生产消息

@Transactional
    @Service("producer")
    public class ProducerImpl implements Producer{ 
    	@Autowired
    	RabbitTemplate rabbitTemplate;
    	public void sendMail(String queue,Mail mail) { 
    		rabbitTemplate.setQueue(queue);
    		rabbitTemplate.convertAndSend(queue,mail);
    	}
    
    }

同时使用 rabbitTeplate 的 convertAndSend 方法相队列消费消息

package boot.spring.service.impl;
    
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    import boot.spring.po.Mail;
    import boot.spring.service.Publisher;
    
    
    @Service("publisher")
    public class PublisherImpl implements Publisher{ 
    	@Autowired
    	RabbitTemplate rabbitTemplate;
    
    	public void publishMail(Mail mail) { 
    		rabbitTemplate.convertAndSend("fanout", "", mail);
    	}
    
    	public void senddirectMail(Mail mail, String routingkey) { 
    		rabbitTemplate.convertAndSend("direct", routingkey, mail);
    	}
    
    	public void sendtopicMail(Mail mail, String routingkey) { 
    		rabbitTemplate.convertAndSend("mytopic", routingkey, mail);
    	}
    
    	
    	
    }

我们可以使用前端请求的方式通知rabbitmq 生产和发送消息，如何通知消息队列去监听消息呢 可以使用 @RabbitListener

@Component
    public class SubscribeListener1 { 
    	
    	@Autowired
    	LoginService loginService;
    	
    	@RabbitListener(queues = "queue1")
    	public void subscribe(Mail mail) throws IOException { 
    		System.out.println("订阅者1收到消息"+mail.toString());
    		loginService.sendMessage();
    	}
    }

@Component
    public class SubscribeListener2 { 
    	
    	@Autowired
    	LoginService loginService;
    	
    	@RabbitListener(queues = "queue2")
    	public void subscribe(Mail mail) throws IOException { 
    		System.out.println("订阅者2收到消息"+mail.toString());
    		loginService.sendMail();
    		
    	}
    }

#### 使用注解 ####

#### 功能说明 ####

#### 代码实现 ####