SpringBoot 使用(四): 消息队列使用

2016-01-13 本文已影响22887人龙白一梦

一. 概述

在整套的微服务架构中, 消息队列是不可或缺的部分, 它能够起到线程内同步或者异步调用无法达到的作用,优缺点分别是:
优点：

解耦
i. 只依赖消息的格式, 而不依赖发送者的ip和端口
ii. 多消费者的情况下, 发送者不需要关注消费者的任何信息
路由
不能互相访问的网络之间可以消息队列实现访问, 可以减少对现有网络的修改。
消息可靠性
当消费者发生故障时, 消息可以被有效保存下来, 等待恢复后继续访问.
异步调用
发送者异步发送消息, 不等待消息ack,不会对发送者本身产生响应速度的影响, 当然异步调用也是可以实现的。
方便扩展
集群部署消息队列, 当流量增大和减小是可以通过调整部署来实现和发送方, 消费方无关。

缺点：
多出一个环节,需要保证消息队列的可用性。

二. 常用消息队列

目前常用的消息队列大概有三种类型,RabbitMQ等AMQP系列, Kafka, Redis等kev value系列,它们的使用场景分别是：
1.RabbitMQ：相对重量级高并发的情况，比如数据的异步处理任务的串行执行等.
2.Kafka：基于Pull的模式来处理,具体很高的吞吐量,一般用来进行日志的存储和收集.
3.Redis: 轻量级高并发，实时性要求高的情况，比如缓存，秒杀，及时的数据分析(ELK日志分析框架,使用的就是Redis).

三. SpingBoot 集成消息队列

在SpingBoot中对这个三种都有支持

Redis
请参照其他作者的文章 http://www.jianshu.com/p/a2ab17707eff

RabbitMQ
i. 配置

 <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
 </dependency>

ii.实现方式 RabbitMQDemoConfigration.java

import org.springframework.amqp.core.;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/*

RabbitMQDemo
@author: sunjie
@date: 16/01/10
*/
@Configuration
@SuppressWarnings("SpringJavaAutowiringInspection")
public class RabbitMQDemoConfigration {
@Bean
public CachingConnectionFactory myConnectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
connectionFactory.setHost("192.168.1.1");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/myHost");
return connectionFactory;
}

// 生成CachingConnectionFactory 也可以使用下面的方式,在application.properties
// 中定义好属性即可

// @Autowired
// ConnectionFactory connectionFactory;
@Bean
public DirectExchange myExchange() {
return new DirectExchange("myExchangeDemo", true, false);
}

@Bean
public Queue myQueue() {
return new Queue("myQueueDemo", true);
}

@Bean
public Binding myExchangeBinding(@Qualifier("myExchange") DirectExchange directExchange,
@Qualifier("myQueue") Queue queue) {
return BindingBuilder.bind(queue).to(directExchange).with("routeDemo");
}

@Bean
public RabbitTemplate myExchangeTemlate() {
RabbitTemplate r = new RabbitTemplate(myConnectionFactory());
r.setExchange("myExchangeDemo");
r.setRoutingKey("routeDemo");
return r;
}

/**
- 发送消息,工业使用需要自己做个性化实现
  */
  @Bean
  public void sendMessage(RabbitTemplate myExchangeTemlate) {
  String string = "Hello RabbitmQ";
  myExchangeTemlate.convertAndSend(string);
  }
/**
- 接受消息,工业使用时需要在监听类中实现process逻辑
  */
  @RabbitListener(queues = "myQueueDemo")
  public void process(Message message) {
  System.out.println("__________" + message.getBody().toString() + "__________");
  try {
  this.wait(1000);
  } catch (InterruptedException e) {
  e.printStackTrace();
  }
  }

Kafka 使用
i. pom.xml 配置

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka_2.10</artifactId>
 </dependency>
 <dependency>
   <groupId>org.springframework.integration</groupId>
   <artifactId>spring-integration-kafka</artifactId>
 </dependency>

ii. spring-integration-kafka.xml 也可以研究通过bean方式来实现

<?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:int="http://www.springframework.org/schema/integration"
       xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
       xmlns:task="http://www.springframework.org/schema/task"
       xsi:schemaLocation="http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
    http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
    http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
    http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd">

    <int:channel id="inputToKafka">
        <int:queue/>
    </int:channel>

    <int-kafka:outbound-channel-adapter
            id="kafkaOutboundChannelAdapter"
            kafka-producer-context-ref="kafkaProducerContext"
            channel="inputToKafka">
        <int:poller fixed-delay="1000" time-unit="MILLISECONDS" receive-timeout="0" task-executor="taskExecutor"/>
    </int-kafka:outbound-channel-adapter>

    <task:executor id="taskExecutor" pool-size="5" keep-alive="120" queue-capacity="50000"/>

    <int-kafka:producer-context id="kafkaProducerContext">
        <int-kafka:producer-configurations>
            <int-kafka:producer-configuration broker-list="192.168.2.2:9092"
                                              key-class-type="java.lang.String"
                                              value-class-type="java.lang.String"
                                              topic="1_service"
                                              value-encoder="kafkaEncoder"
                                              key-encoder="kafkaEncoder"/>
        </int-kafka:producer-configurations>
    </int-kafka:producer-context>

    <bean id="kafkaEncoder" class="org.springframework.integration.kafka.serializer.common.StringEncoder"/>
</beans>

iii. 客户端实现代码：

  @Autowired
    MessageChannel             inputToKafka;

    String value = "Hello Kafka";
    Message<String> mess = MessageBuilder.withPayload(value)
            .setHeader(KafkaHeaders.TOPIC, appSetting.getStrValue("topic")).build();
    inputToKafka.send(mess);

四. 后记

消息中间件有很多,实际使用中往往是根据架构师的技术栈相关,做到了解使用场景和基本原理,在项目中提升自己细节能力,做到个人和公司共同成长,才是好的方式.

SpringBoot 使用(四): 消息队列使用

一. 概述

二. 常用消息队列

三. SpingBoot 集成消息队列

四. 后记

猜你喜欢

热点阅读