Kafka分区分配策略

Kafka分区分配策略

我们知道每个Topic会分配为很多partitions,Producers会将数据分配到每个partitions中,然后消费者Consumers从partitions中获取数据消费,那么Producers是如何将数据分到partitions中?Consumers又怎么知道从哪个partitions中消费数据?

生产者往Topic写数据

我们从product.send方法入手,看看里面的具体实现,可以看到在调用send方法时,其内部是调用了doSend方法,在doSend方法中有一个获取partitions的方法

int partition = partition(record, serializedKey, serializedValue, cluster);

private int partition(ProducerRecord<K, V> record, byte[] serializedKey, byte[] serializedValue, Cluster cluster) {
    Integer partition = record.partition();
    return partition != null ?
            partition :
            partitioner.partition(
                    record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
}

从上面代码中,首先先选择配置的分区,如果没有配置则使用默认的分区,即使用了DefaultPartitioner中的partition方法

public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
    List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
    int numPartitions = partitions.size();
    if (keyBytes == null) {
        int nextValue = nextValue(topic);
        List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
        if (availablePartitions.size() > 0) {
            int part = Utils.toPositive(nextValue) % availablePartitions.size();
            return availablePartitions.get(part).partition();
        } else {
            // 没有可用的分区,则给一个不可用分区
            return Utils.toPositive(nextValue) % numPartitions;
        }
    } else {
        // 根据key的hash值和分区数取模
        return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
    }
}

上面代码中先会根据topic获取所有的分区
1,如果key为null,则通过先产生随机数,之后在该数上自增的方式产生一个数nextValue,如果存在可用分区,将nextValue转为正数之后对可用分区进行取模操作,如果不存在可用分区,则将nextValue对总分区数进行取模操作

2,如果key不为空,就先获取key的hash值,然后和分区数进行取模操作

消费者从Topic读数据

kafka默认对消费分区指定了两种策略,分别为Range策略(org.apache.kafka.clients.consumer.RangeAssignor)和RoundRobin策略(org.apache.kafka.clients.consumer.RoundRobinAssignor),它们都实现了PartitionAssignor接口

Range策略

比如有10个分区,分别为P1、P2、P3、P4、P5、P6、P7、P8、P9、P10,三个消费者C1、C2、C3,消费如下图:

C4F4C4E974548380CE4D5274D30F20B9

我们来看看源代码:

@Override
public Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic,
                                                Map<String, Subscription> subscriptions) {
    // 得到topic和订阅的消费者集合信息,例如{t1:[c1,c2,c3], t2:[c1,c2,c3,c4]}                                            
    Map<String, List<String>> consumersPerTopic = consumersPerTopic(subscriptions);
    Map<String, List<TopicPartition>> assignment = new HashMap<>();
    // 将consumersPerTopic信息转换为assignment,memberId就是消费者client.id+uuid(kafka在client.id上追加的)
    for (String memberId : subscriptions.keySet())
        assignment.put(memberId, new ArrayList<TopicPartition>());

    // 遍历每个Topic,获取所有的订阅消费者
    for (Map.Entry<String, List<String>> topicEntry : consumersPerTopic.entrySet()) {
        String topic = topicEntry.getKey();
        List<String> consumersForTopic = topicEntry.getValue();

        Integer numPartitionsForTopic = partitionsPerTopic.get(topic);
        // 如果Topic没有分区,则调过
        if (numPartitionsForTopic == null)
            continue;

         // 将Topic的订阅者根据字典排序
        Collections.sort(consumersForTopic);
         // 总分区数/订阅者的数量 得到每个订阅者应该分配分区数
        int numPartitionsPerConsumer = numPartitionsForTopic / consumersForTopic.size();
        // 无法整除的剩余分区数量
        int consumersWithExtraPartition = numPartitionsForTopic % consumersForTopic.size();

        List<TopicPartition> partitions = AbstractPartitionAssignor.partitions(topic, numPartitionsForTopic);
        //遍历所有的消费者
        for (int i = 0, n = consumersForTopic.size(); i < n; i++) {
              //分配到的分区的开始位置
            int start = numPartitionsPerConsumer * i + Math.min(i, consumersWithExtraPartition);
            // 分配到的分区数量(整除分配到的分区数量,加上1个无法整除分配到的分区--如果有资格分配到这个分区的话。判断是否有资格分配到这个分区:如果整除后余数为m,那么排序后的消费者集合中前m个消费者都能分配到一个额外的分区)
            int length = numPartitionsPerConsumer + (i + 1 > consumersWithExtraPartition ? 0 : 1);
            //给消费者分配分区
            assignment.get(consumersForTopic.get(i)).addAll(partitions.subList(start, start + length));
        }
    }
    return assignment;
}

上面的代码添加了注释很清楚的展现了range的实现,对应上面的例子,如果有4个消费者C1、C2、C3、C4,那么根据上面的算法:

C1 -> [P1,P2,P3] ,C2 -> [P4,P5,P6] ,C3 -> [P7,P8] C4 -> [P9,P10] 。取余多出来的两个分区,由最前n个消费者来消费

RoundRobin策略

将主题的所有分区依次分配给消费者,比如有两个Topic:T1[P1,P2,P3,P4],T2[P5,P6,P7,P8,P9,P10],若C1、C2订阅了T1,C2、C3订阅了T2,那么C1将消费T1[P1,P3],C2将消费T1[P2,P4,P6,P8,P10],C3将消费T2[P5,P7,P9],如下图:

A8E30AE2B2B3245CCEB5102A9F1B8470

@Override
public Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic,
                                                Map<String, Subscription> subscriptions) {
    Map<String, List<TopicPartition>> assignment = new HashMap<>();
    for (String memberId : subscriptions.keySet())
        assignment.put(memberId, new ArrayList<TopicPartition>());
    // 将消费集合先按字典排序,构建成一个环形迭代器
    CircularIterator<String> assigner = new CircularIterator<>(Utils.sorted(subscriptions.keySet()));
   // 按Topic的名称排序,得到Topic下的所有分区
    for (TopicPartition partition : allPartitionsSorted(partitionsPerTopic, subscriptions)) {
        final String topic = partition.topic();

        while (!subscriptions.get(assigner.peek()).topics().contains(topic))
            assigner.next();
        // 给消费者分配分区,并轮询到下一个消费者
        assignment.get(assigner.next()).add(partition);
    }
    return assignment;
}

/**
 * 根据消费者得到订阅的Topic下的所有分区
 * Topic按名称字典排序
 */
public List<TopicPartition> allPartitionsSorted(Map<String, Integer> partitionsPerTopic,
                                                Map<String, Subscription> subscriptions) {
    SortedSet<String> topics = new TreeSet<>();
    for (Subscription subscription : subscriptions.values())
        topics.addAll(subscription.topics());

    List<TopicPartition> allPartitions = new ArrayList<>();
    for (String topic : topics) {
        Integer numPartitionsForTopic = partitionsPerTopic.get(topic);
        if (numPartitionsForTopic != null)
            allPartitions.addAll(AbstractPartitionAssignor.partitions(topic, numPartitionsForTopic));
    }
    return allPartitions;
}

 

使用Sharing-JDBC实现分表

使用Sharing-JDBC实现分表

Sharing-JDBC介绍

在创建数据库时,我们最先考虑的是按模块对数据库进行划为,但即使这样,单表数据量还是出现大数量的情况,Sharing-JDBC可以对表进行水平切分,将数据均分到不同表中

通过日期水平切分

目前我们航班动态数据全球每天航班20多万,考虑到我们业务场景,用户都是通过航班号+日期来查询一个航班,所以我们采取使用日期来水平分表

添加依赖

<dependency>
	<groupId>com.dangdang</groupId>
	<artifactId>sharding-jdbc-core</artifactId>
	<version>1.5.4.1</version>
</dependency>
<dependency>
	<groupId>com.dangdang</groupId>
	<artifactId>sharding-jdbc-config-spring</artifactId>
	<version>1.5.4.1</version>
</dependency>
<!--mysql flight数据源 -->
<bean id="mysqlDataSource" class="com.zaxxer.hikari.HikariDataSource" destroy-method="close">
...
</bean>

<!--shared jdbc -->

<rdb:strategy id="tableShardingStrategy" sharding-columns="local_date" algorithm-class="com.huoli.songshan.sharding.FlyDateTableShardingAlgorithm" />
<rdb:data-source id="shardingDataSource">
	<rdb:sharding-rule data-sources="mysqlDataSource" default-data-source="mysqlDataSource">
		<rdb:table-rules>
			<rdb:table-rule logic-table="flight_info" actual-tables="flight_info_${2012..2020}${['01','02','03','04','05','06','07','08','09','10','11','12']}${0..3}${0..9}"
				table-strategy="tableShardingStrategy" />
		</rdb:table-rules>
		<rdb:default-database-strategy sharding-columns="none" algorithm-class="com.dangdang.ddframe.rdb.sharding.api.strategy.database.NoneDatabaseShardingAlgorithm" />
	</rdb:sharding-rule>
	<rdb:props>
		<prop key="sql.show">false</prop>
	</rdb:props>
</rdb:data-source>

先配置一个dataSource数据源,这里我们使用的是hikari,再通过shardingDataSource对dataSource进行包装,按照表中localdate字段对表进行拆分,即表名依次为flightinfo20120101、flightinfo20120102,到flightinfo20201231,下面我们实现根据日期localdate映射到对应表上

public final class FlyDateTableShardingAlgorithm implements SingleKeyTableShardingAlgorithm<Date> {
    private DateTimeFormatter dt = DateTimeFormat.forPattern("yyyyMMdd");

    @Override
    public String doEqualSharding(Collection<String> availableTargetNames, ShardingValue<Date> shardingValue) {
        DateTime datetime = new DateTime(shardingValue.getValue());
        String flydate = datetime.toString(dt);
        for (String each : availableTargetNames) {
            if (each.endsWith(flydate)) {
                return each;
            }
        }
        throw new IllegalArgumentException();
    }

    @Override
    public Collection<String> doInSharding(Collection<String> availableTargetNames, ShardingValue<Date> shardingValue) {
        Collection<String> result = new LinkedHashSet<>(availableTargetNames.size());
        for (Date value : shardingValue.getValues()) {
            DateTime datetime = new DateTime(value);
            String flydate = datetime.toString(dt);
            for (String tableName : availableTargetNames) {
                if (tableName.endsWith(flydate)) {
                    result.add(tableName);
                }
            }
        }
        return result;
    }

    @Override
    public Collection<String> doBetweenSharding(Collection<String> availableTargetNames,
            ShardingValue<Date> shardingValue) {
        Collection<String> result = new LinkedHashSet<>(availableTargetNames.size());
        Range<Date> range = (Range<Date>) shardingValue.getValueRange();
        for (Date value = range.lowerEndpoint(); value.before(range.upperEndpoint()) || value.equals(range.upperEndpoint()); value = addDays(value, 1)) {
            DateTime datetime = new DateTime(value);
            String flydate = datetime.toString(dt);
            for (String each : availableTargetNames) {
                if (each.endsWith(flydate)) {
                    result.add(each);
                }
            }
        }
        return result;
    }

    private Date addDays(Date date, int days) {
        return new Date(new DateTime(new DateTime(date).plusDays(days)).toDate().getTime());
    }
}

这里实现了SQL的equal、in、between方法,即通过使用select * from flight_info where local_date=’2019-01-01′,会自动映射flight_info_20190101表中查询。

通过用户ID哈希取模进行拆分

在用户订阅航班动态数据后,我们需要保存用户的订阅数据,以便后期推送动态消息给用户,用户可以在登陆的情况下订阅航班,也可以在未登陆的情况下订阅航班,用户登陆后可以拿到用户的phoneId(设备ID)和userId(用户ID),用户未登陆只能获取到用户的phoneId(设备ID),所以我们区分两种情况,使用两种表来存放用户信息trip_subscribe_nologin和trip_subscribe_login,用户未登陆根据phoneId来分表,用户登陆的情况根据userId来分表

和之前一样,我们需要先配置一个分表策略

<!-- trip_subscribe_nonlogin分表策略 -->
<rdb:strategy id="subscribeNonLoginTableShardingStrategy" sharding-columns="uid" algorithm-class="com.huoli.trip.dao.sharding.SubscribeNonLoginTableShardingAlgorithm" />
<!-- trip_subscribe_login分表策略 -->
<rdb:strategy id="subscribeLoginTableShardingStrategy" sharding-columns="user_id" algorithm-class="com.huoli.trip.dao.sharding.SubscribeLoginTableShardingAlgorithm" />

<!-- 分表配置配置 -->
<rdb:data-source id="shardingDataSource">
	<rdb:sharding-rule data-sources="mysqlDataSource">
		<rdb:table-rules>
			<rdb:table-rule logic-table="trip_subscribe_nonlogin" table-strategy="subscribeNonLoginTableShardingStrategy" actual-tables="trip_subscribe_nonlogin_${0..63}">
				<rdb:generate-key-column column-name="id" column-key-generator-class="com.huoli.trip.dao.sharding.TripKeyGenerator" />
			</rdb:table-rule>
			<rdb:table-rule logic-table="trip_subscribe_login" table-strategy="subscribeLoginTableShardingStrategy" actual-tables="trip_subscribe_login_${0..63}">
				<rdb:generate-key-column column-name="id" column-key-generator-class="com.huoli.trip.dao.sharding.TripKeyGenerator" />
			</rdb:table-rule>
		</rdb:table-rules>
		<rdb:default-database-strategy sharding-columns="none" algorithm-class="com.dangdang.ddframe.rdb.sharding.api.strategy.database.NoneDatabaseShardingAlgorithm" />
	</rdb:sharding-rule>
	<rdb:props>
		<prop key="sql.show">true</prop>
	</rdb:props>
</rdb:data-source>

这里是根据phoneId、userId分别分成了64张表,即trip_subscribe_nologin有64张表(trip_subscribe_nologin_1,trip_subscribe_nologin_2 …),trip_subscribe_login有64张表(trip_subscribe_login_1,trip_subscribe_login_2 …),具体代码实现

public interface Shard<T> {

    /** 根据参数计算分片标识 */
    public T calculateShard(Object... args);
}

public class SubscribeNonLoginTableShardingAlgorithm implements SingleKeyTableShardingAlgorithm<String>, Shard<String> {

    /** 分片数量 */
    private final int shardNum = 64;

    public String doEqualSharding(Collection<String> availableTargetNames, ShardingValue<String> shardingValue) {
        String shard = calculateShard(shardingValue.getValue());
        for (String tableName : availableTargetNames) {
            if (tableName.endsWith("_" + shard)) {
                return tableName;
            }
        }

        throw new IllegalArgumentException("未找到该表:trip_subscribe_nonlogin_" + shard);
    }

    public Collection<String> doInSharding(Collection<String> availableTargetNames, ShardingValue<String> shardingValue) {
        Collection<String> result = new LinkedHashSet<String>(availableTargetNames.size());
        for (String value : shardingValue.getValues()) {
            String shard = calculateShard(value);
            for (String tableName : availableTargetNames) {
                if (tableName.endsWith("_" + shard)) {
                    result.add(tableName);
                }
            }
        }
        return result;
    }

    public Collection<String> doBetweenSharding(Collection<String> availableTargetNames, ShardingValue<String> shardingValue) {
        // 不会出现between两个uid之间的查询需求,所以无需实现该方法
        return new LinkedHashSet<String>(availableTargetNames.size());
    }

    public String calculateShard(Object... args) {
        String phoneId = (String) args[0];
        return "" + (phoneId.hashCode() & 0x7fffffff) % shardNum;
    }
}

public class SubscribeLoginTableShardingAlgorithm implements SingleKeyTableShardingAlgorithm<String>, Shard<String> {

    /** 分片数量 */
    private final int shardNum = 64;

    /**
     * doEqualSharding/doInSharding同SubscribeNonLoginTableShardingAlgorithm
     */

    public String calculateShard(Object... args) {
        String userId = (String) args[0];
        return "" + (userId.hashCode() & 0x7fffffff) % shardNum;
    }
}

这里主要介绍下calculateShard方法,首先是先拿到了phoneId或userId,获取hashCode,然后和0x7fffffff进行&运算,0x7fffffff表示long的最大值,这一步是为了保证得到的index的第一位为0,也就是为了得到一个正数。因为有符号数第一位0代表正数,1代表负数,然后和分片数shardNum,使表数据分布在shardNum内

另外这里还用到了id的生成策略,即生成一个全局的自增ID,sharing-jdbc自带了DefaultKeyGenerator生成器可以实现

public class TripKeyGenerator implements KeyGenerator {
    private static Logger logger = LoggerFactory.getLogger(TripKeyGenerator.class);

    private DefaultKeyGenerator defaultKeyGenerator;

    public TripKeyGenerator() {
        defaultKeyGenerator = new DefaultKeyGenerator();
    }

    static {
        /** 从配置中获取workId */
        DefaultKeyGenerator.setWorkerId(workerId);
    }

    @Override
    public synchronized Number generateKey() {
        return defaultKeyGenerator.generateKey();
    }

}

通过twitter的snowlflake也可以生成唯一ID,具体可以参考:这里

 

Elasticsearch排序(三)

Elasticsearch排序(三)

Elasticsearch中排序是通过设置score字段来进行排序,score是一个浮点类型,所以我们对靠前的数据设置一个较大的值,然后根据倒序排列。

同样对于航班号查询,有时候需要根据航班号,日期查询出所有的航班号,用户会输入185,2018-12-01,这时需要返回所有相关航司的航班号。
比如匹配的航班有:CA1858,B6185,QF185,ZH1858,对于精确匹配,B6185、QF185要排在CA1858、ZH1858之前,对于非精确排序,CA航司要排在ZH航司前面

创建一个索引

之前我们对航班号创建了一个索引,用户可以输入CA1,出现CA1*的结果,现在用户输入185,是纯数字型的,所以我们需要单独一个字段(flightNum)保存这个航班数字,然后对它创建索引

PUT flight
{
  "settings": {
    "analysis": {
      "analyzer": {
        "flightNoAndNumAnalyzer": {
          "tokenizer": "flightNoAndNumTokenizer"
        }
      },
      "tokenizer": {
        "flightNoAndNumTokenizer": {
          "type": "edge_ngram",
          "min_gram": 3,
          "max_gram": 8,
          "token_chars": ["letter","digit"]
        }
      }
    }
  },
  "mappings": {       
      "dynamic": {           
          "properties": {   
              "flightNo": {
                 "type": "text",
                 "analyzer" : "flightNoAndNumAnalyzer"                          
              },
              "flightNum": {
                 "type": "text",
                 "analyzer" : "flightNoAndNumAnalyzer"                          
              }
          }
      }
  }

}

如果是航班数字查询,我们只需要根据航司来设定优先级,同时设定航班数字全匹配的优先级最高,然后查询的时候再通过constant_score统一来设置boost参数

GET flight/dynamic/_search
{
  "query": {
    "bool": {
      "must": [
        {"prefix": {
          "flightNum": {
            "value": "185"
          }
        }}
      ], 
      "should": [
        {
          "constant_score": {
            "filter": {
              "term": {
                "airline": "CA"
              }
            },
            "boost": 4
          }
        },
        {
          "constant_score": {
            "filter": {
              "term": {
                "airline": "ZH"
              }
            },
            "boost": 3
          }
        },
        {
          "constant_score": {
            "filter": {
              "term": {
                "airline": "B6"
              }
            },
            "boost": 2
          }
        },
        {
          "constant_score": {
            "filter": {
              "term": {
                "airline": "QF"
              }
            },
            "boost": 1
          }
        },
        {
          "constant_score": {
            "filter": {
              "term": {
                "flightNum": "185"
              }
            },
            "boost": 10
          }
        }
      ]
      }
    }
}

可以看到CA航司比B6高,但是当查询flightNum为185的时候,评分设置为最高,所以B6186会优先显示,其次是QF185,最后是CA1858和ZH1858 具体代码实现:

//航班数字优先级设置
BoolQueryBuilder qb = QueryBuilders.boolQuery()
                    .must(prefixQuery("flightNum", keyword))
                    .should(constantScoreQuery(matchQuery("flightNum", keyword)).boost(10f))

//航司优先级设置
airNumer.put("CA", 4f);
airNumer.put("ZH", 3f);
airNumer.put("B6", 4f);
airNumer.put("QF", 1f);
for (String airline : airNumer.keySet()) {
    queryBuilder.should(constantScoreQuery(matchQuery("airline", airline)).boost(airNumer.get(airline)));
}

 

 

 

Elasticsearch创建索引及数据(二)

Elasticsearch创建索引及数据(二)

第一篇介绍了Elasticsearch安装及基本用法,下面我们自己来创建一个索引,并写入一些数据

创建索引

基本语法

PUT /my_index
{
    "settings": { ... any settings ... },
    "mappings": {
        "type_one": { ... any mappings ... },
        "type_two": { ... any mappings ... },
        ...
    }

在创建索引的时候,我们需要设置索引被存放的分片数量、分析器、类型设置,分片数量、分析器通过settings来设置,类型映射通过mappings来设置,Elasticsearch默认创建索引是5个分片,1个副本,可通过settings来修改它,这里我们就遵循默认值不动,如下:

PUT flight
{
    "settings" : {
        "index" : {
            "number_of_shards" : 5, 
            "number_of_replicas" : 1 
        }
    }
}
{
  "acknowledged": true,
  "shards_acknowledged": true,
  "index": "flight"
}

创建一个flight索引,大括号后面为默认配置,如果不想修改它,都可以去掉,然后在新建一个type为dynamic,并写入数据

PUT /flight/dynamic/1
{
    "flightNo":"CA1858",
    "flightDate":"2018-12-01",
    "depCode":"SHA",
    "arrCode":"PEK",
    "state": "到达",
    "subState": "",
    "depPlanTime":"2018-12-01 07:45",
    "arrPlanTime":"2018-12-01 10:10",
    "depReadyTime":"2018-12-01 07:45",
    "arrReadyTime":"2018-12-01 09:45",
    "depTime":"2018-12-01 07:54",
    "arrTime":"2018-12-01 07:46",
    "distance":1076, 
    "tailNo":"B2487", 
    "depTerm":"T2", 
    "arrTerm":"T3", 
    "gate":"48", 
    "luggage":"32"
}

创建一条数据后,ES会默认会对所有字段添加索引,当然也可以不指定ID,ES会默认生成一个ID,自动生成的ID有22个字符长,类似:wM0OSFhDQXGZAWDf0-drSA

{
  "_index": "flight",
  "_type": "dynamic",
  "_id": "1",
  "_version": 1,
  "result": "created",
  "_shards": {
    "total": 2,
    "successful": 1,
    "failed": 0
  },
  "_seq_no": 0,
  "_primary_term": 1
}

下面我们来查询数据,ES默认情况下是禁用了source,即我们查询时,不会返回source里面的内容,只会返回数据ID,我们可以指定需要返回的_srouce字段

GET flight/dynamic/_search
{
    "query":   { "match_all": {}},
    "_source": [ "flightNo", "flightDate"]
}
{
  "took": 2,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": 1,
    "max_score": 1,
    "hits": [
      {
        "_index": "flight",
        "_type": "dynamic",
        "_id": "1",
        "_score": 1,
        "_source": {
          "flightNo": "CA1858",
          "flightDate": "2018-12-01"
        }
      }
    ]
  }
}

配置分析器

分析器介绍

分析器主要是将一块文本分成适合于倒排索引的独立 词条,通过这些词条我们可以搜索到指定的文档

分析器主要有以下几个功能:
1,字符过滤:通过整理字符串,如去掉HTML,将&转换为and
2,分词器:将字符串分成单个词条,比如通过空格或者标点符号来拆分
3,Token过滤:比如将Quick这种词条统一转换为小写,删除a,and,the这些无用词,增加近义词条(如:jump和leap这种同义词)

比如我们有一个航班号CA1858,我们需要通过输入CA18就可以返回对应的结果,那么我们就需要对其进行分词,如果直接使用下面的查询是获取不到数据的

GET flight/dynamic/_search
{
  "query": {
    "match": {
      "flightNo": "CA18"
    }
  }
}

我们可以查看ES对flightNo的分词情况

POST flight/_analyze
{
  "field": "flightNo",
  "text": "CA1858"
}
{
  "tokens": [
    {
      "token": "ca1858",
      "start_offset": 0,
      "end_offset": 6,
      "type": "<ALPHANUM>",
      "position": 0
    }
  ]
}

可以看到ES是对航班号转换为小写后直接进行了倒排索引,没有进行分词,直接查询CA18肯定搜索不到,下面我们自定义一个分析器

PUT flight
{
  "settings": {
    "analysis": {
      "analyzer": {
        "flightNoAnalyzer": {
          "tokenizer": "flightNoTokenizer"
        }
      },
      "tokenizer": {
        "flightNoTokenizer": {
          "type": "edge_ngram",
          "min_gram": 4,
          "max_gram": 8,
          "token_chars": ["letter","digit"]
        }
      }
    }
  },
  "mappings": {       
      "dynamic": {           
          "properties": {   
              "flightNo": {
                 "type": "text",
                 "analyzer" : "flightNoAnalyzer"                          
              }
          }
      }
  }

}

edgengram为ES自带的分词器,ES自带了8种分析器,具体可以查看官方文档,我们自定义了一个analyzer,使用edgengram来进行分词,字符长度从4到8,然后将自定义的分析器映射到flightNo字段上,之后我们使用查询CA18就可以获取到结果了

 

Elasticsearch安装及介绍(一)

Elasticsearch安装及介绍

安装Elasticsearch

Elasticsearch的安装很简单,直接下载官方包,运行即可

curl -L -O http://download.elasticsearch.org/PATH/TO/VERSION.zip 
unzip elasticsearch-$VERSION.zip
cd  elasticsearch-$VERSION
./bin/elasticsearch

使用curl ‘http://localhost:9200/?pretty’获取数据

{
  "name" : "node-0",
  "cluster_name" : "dongtai-es",
  "cluster_uuid" : "aIah5IcqS1y9qiJ4LphAbA",
  "version" : {
    "number" : "6.1.2",
    "build_hash" : "5b1fea5",
    "build_date" : "2018-01-10T02:35:59.208Z",
    "build_snapshot" : false,
    "lucene_version" : "7.1.0",
    "minimum_wire_compatibility_version" : "5.6.0",
    "minimum_index_compatibility_version" : "5.0.0"
  },
  "tagline" : "You Know, for Search"
}

基本概念介绍

Elasticsearch本质是一个分布式数据库,每个服务器上称为一个Elastic实例,这一组实例构成了ES集群

Document

ES中存储数据记录称为Document,Document使用JSON格式表示,同一个 Index 里面的Document没有要求有相同的结构,但最好一致,这样能提高效率,如下:

{
  "first_name": "Jane2",
  "last_name": "Smith",
  "age": 32,
  "about": "I like to collect rock albums",
  "interests": "music"
}

Index

ES文档的索引,定义了文档放在哪里,你可以理解为对于数据库名

Type

文档表示的对象类别,可以理解为数据库中的表,比如我们有一个机票数据,可以按舱位来分组(头等舱,经济舱),根据规则,Elastic 6.x 版只允许每个 Index 包含一个 Type,7.x 版将会彻底移除 Type

ID

文档的唯一标识,可以理解为表中的一条记录,根据index、type、_ID可以确定一个文档

基本用法

添加一个文档

PUT index/employee/6
{
  "first_name": "chuan",
  "last_name": "zhang",
  "age": 32,
  "about": "hi I'm hai na bai chuan",
  "interests": "read book"
}

上面PUT后面分别对应 index、Type、Id

获取文档

获取所有文档

ES默认会返回10条,可以指定size来改变

GET index/employee/_search
{
  "size": 3
}
{
  "took": 1,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": 9,
    "max_score": 1,
    "hits": [
      {
        "_index": "index",
        "_type": "employee",
        "_id": "5",
        "_score": 1,
        "_source": {
          "first_name": "Jane1",
          "last_name": "Smith",
          "age": 32,
          "about": "I like to collect rock albums",
          "interests": "music"
        }
      },
      {
        "_index": "index",
        "_type": "employee",
        "_id": "10",
        "_score": 1,
        "_source": {
          "first_name": "chuan",
          "last_name": "Smith",
          "age": 32,
          "about": "I like to collect rock albums",
          "interests": "music"
        }
      }
    ]
  }
}

按条件匹配

比如我需要查找firstname是’chuan‘的所有文档,如果要搜索firstname中多个关键字,中间用空格隔开就可以了,比如”firstname”: “chuan smith” 表示查询firstname包含chuan或smith的

GET index/employee/_search
{
  "query": {
    "match": {
            "first_name": "chuan"
        }
  }
}
{
  "took": 4,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": 1,
    "max_score": 0.6931472,
    "hits": [
      {
        "_index": "index",
        "_type": "employee",
        "_id": "6",
        "_score": 0.6931472,
        "_source": {
          "first_name": "chuan",
          "last_name": "zhang",
          "age": 32,
          "about": "hi I'm hai na bai chuan",
          "interests": "read book"
        }
      }
    ]
  }
}

如果要使用and查询,比如我要查询firstname为chuan,lastname为chuan的文档记录

GET index/employee/_search
{
  "query": {
    "bool": {
      "must": [
        { "match": { "first_name": "chuan" } },
        { "match": { "last_name": "zhang" } }
      ]
    }
  }
}

如果要根据范围查询或过滤,类似MySQL中’>’,或 ‘!=’操作,比如需要查询上面条件中,年龄大于30的文档

{
  "query": {
    "bool": {
      "must": [
        { "match": { "first_name": "chuan" } },
        { "match": { "last_name": "zhang" } }
      ]
      "filter": {
        "range" : {
            "age" : { "gt" : 30 } 
        }
      }
    }
  }
}

更新文档

更新记录使用PUT请求,重新发送一次数据,ES会根据id来修改,如果我需要更新id为6的about字段

PUT index/employee/6
{
  "first_name": "chuan",
  "last_name": "zhang",
  "age": 32,
  "about": "i modify my about",
  "interests": "read book"
}
{
  "_index": "index",
  "_type": "employee",
  "_id": "6",
  "_version": 2,
  "result": "updated",
  "_shards": {
    "total": 2,
    "successful": 1,
    "failed": 0
  },
  "_seq_no": 12,
  "_primary_term": 2
}

_version为版本号,由原来的1变成了2,result为”update“,以上为全量更新,如果需要局部更新,可以使用POST请求

POST index/employee/6/_update
{
  "doc":{
    "about": "i modify my about"
  }
}

删除文档

如下,我需要删除id为6的文档

DELETE index/employee/6
{
  "found" :    true,
  "_index" :   "index",
  "_type" :    "employee",
  "_id" :      "6",
  "_version" : 3
}

可以看到version也增加了1,如果对于的id没有找到,found将返回false

数据分析

ES同样有类似MySQL group by的用法,ES称为聚合,比如我想统计employee中最受欢迎的兴趣,注意:ES在5.x之后对排序、聚合操作用单独的数据结构(fielddata)缓存到内存里了,需要单独开启。

PUT index/_mapping/employee/
{
  "properties": {
    "interests": { 
      "type":     "text",
      "fielddata": true
    }
  }
}

GET index/employee/_search
{
  "aggs": {
    "all_interests": {
      "terms": { "field": "interests" }
    }
  }
}
"aggregations": {
    "all_interests": {
      "doc_count_error_upper_bound": 0,
      "sum_other_doc_count": 0,
      "buckets": [
        {
          "key": "music",
          "doc_count": 3
        },
        {
          "key": "forestry",
          "doc_count": 1
        },
        {
          "key": "read",
          "doc_count": 1
        }
      ]
    }
  }

Spring Boot集成Quartz

Spring Boot集成Quartz

我们常用的定时任务调度有如下几种:
1. 使用JDK自带的类库实现

  • 通过继承TimeTask,使用Time来调度任务
  • 使用ScheduledExecutorService来实现任务调度

2,Spring 自带了任务调度功能,通过使用@Schedule()注解来实现
3,使用Quartz框架,Quartz可以用于在分布式环境下的任务调度

Quartz的基本概念:

1,Job 表示一个工作,要执行的具体内容。此接口中只有一个方法,如下:

void execute(JobExecutionContext context)

2,JobDetail 表示一个具体的可执行的调度程序,Job 是这个可执行程调度程序所要执行的内容,另外 JobDetail 还包含了这个任务调度的方案和策略。
3,Trigger 代表一个调度参数的配置,什么时候去调。
4,Scheduler 代表一个调度容器,一个调度容器中可以注册多个 JobDetail 和 Trigger。当 Trigger 与 JobDetail 组合,就可以被 Scheduler 容器调度了。

配置环境依赖

添加Maven依赖:

    org.springframework.boot
    spring-boot-starter-quartz

修改application.yml文件,添加:

spring:
  datasource:
    name: schedule
    driver-class-name: com.mysql.jdbc.Driver
    url: jdbc:mysql://localhost:3306/schedule?useUnicode=true&characterEncoding=utf-8
    username: root
    password: 123456
  quartz:
    job-store-type: jdbc

quartz支持两种存储方式,一个使用数据库,也就是job-store-type=jdbc,需要你下载quartz提供的表,并导入到你本地,然后配置你的数据库连接,官方下载地址, 另一种是内存方式,job-store-type设置为momery

配置任务

创建一个任务配置,这里的数据可以对应数据库一条记录

public class ScheduleTaskConfig {

    private Integer id;
    private String name;
    private String groupName;
    private String describe;
    private String cron;
    private String classPath;
    private Character isEnabled;
    private String createTime;
    private String updateTime;

    public String getClassName() {
        try {
            Class cls = Class.forName(classPath);
            return cls.getSimpleName();
        } catch (ClassNotFoundException e) {
            log.error("获取Class失败",e);
        }
        return null;
    }

}

创建用于操作Quartz任务创建、暂停、恢复方法

Service
@Slf4j
public class QuartzService {

    @Autowired
    private Scheduler scheduler;

    public void createJob(ScheduleTaskConfig config) {
        Class cls = null;
        try {
            cls = Class.forName(config.getClassPath());
            JobDataMap jobDataMap = new JobDataMap();
            jobDataMap.put("config", config);
            JobDetail jobDetail = JobBuilder.newJob(cls)
                    .withIdentity(config.getName(), config.getGroupName())
                    .withDescription(config.getDescribe())
                    .setJobData(jobDataMap)
                    .storeDurably()
                    .build();

            Trigger trigger = TriggerBuilder.newTrigger()
                    .forJob(jobDetail)
                    .withIdentity(jobDetail.getKey().getName(), GROUP_NAME)
                    .withSchedule(CronScheduleBuilder.cronSchedule(config.getCron()))
                    .build();
            scheduler.scheduleJob(jobDetail, trigger);
        } catch (Exception e) {
            log.error("创建任务失败", e);
        }
    }

    /**
     * 获取任务状态
     * @return
     */
    public String getJobState(ScheduleTaskConfig config) {
        try {
            TriggerKey triggerKey = TriggerKey.triggerKey(config.getName(), config.getGroupName());
            Trigger.TriggerState state = scheduler.getTriggerState(triggerKey);
            return state.name();
        } catch (Exception e) {
            log.error("创建任务失败", e);
        }
        return null;
    }

    /**
     * 判断Job是否存在
     * @param config
     * @return
     */
    public boolean isExistJob(ScheduleTaskConfig config) {
        try {
            TriggerKey triggerKey = TriggerKey.triggerKey(config.getName(), config.getGroupName());
            CronTrigger trigger = (CronTrigger)scheduler.getTrigger(triggerKey);
            if (trigger != null) {
                return true;
            }
        } catch (Exception e) {
            log.error("获取任务trigger失败", e);
        }

        return false;
    }

    /**
     * 更新Job任务
     * @param config
     */
    public void updateJob(ScheduleTaskConfig config) {
        try {
            TriggerKey triggerKey = TriggerKey.triggerKey(config.getName(), config.getGroupName());
            CronTrigger trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
            CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(config.getCron());

            //按新的cronExpression表达式重新构建trigger
            trigger = trigger.getTriggerBuilder().withIdentity(triggerKey).withSchedule(scheduleBuilder).build();
            Trigger.TriggerState triggerState = scheduler.getTriggerState(trigger.getKey());
            // 忽略状态为PAUSED的任务,解决集群环境中在其他机器设置定时任务为PAUSED状态后,集群环境启动另一台主机时定时任务全被唤醒的bug
            if(!triggerState.name().equalsIgnoreCase("PAUSED")){
                //按新的trigger重新设置job执行
                scheduler.rescheduleJob(triggerKey, trigger);
            }
        } catch (Exception e) {
            log.error("更新任务trigger失败", e);
        }

    }

    /**
     * 停止任务
     * @param jobName
     * @param groupName
     * @return
     */
    public boolean pauseJob(String jobName) {
        try {
            JobKey jobKey = JobKey.jobKey(config.getName(), config.getGroupName());
            scheduler.pauseJob(jobKey);
            return true;
        } catch (SchedulerException e) {
            log.error("停止任务失败", e);
        }
        return  false;
    }

    /**
     * 恢复任务
     * @param jobName
     * @param groupName
     * @return
     */
    public boolean resumeJob(String jobName) {
        try {
            JobKey jobKey = JobKey.jobKey(config.getName(), config.getGroupName());
            scheduler.resumeJob(jobKey);
            return true;
        } catch (SchedulerException e) {
            log.error("恢复任务失败", e);
        }
        return  false;
    }

    /**
     * 删除任务
     * @param jobName
     * @param groupName
     * @return
     */
    public boolean deleteJob(String jobName) {
        try {
            JobKey jobKey = JobKey.jobKey(config.getName(), config.getGroupName());
            scheduler.deleteJob(jobKey);
            return true;
        } catch (SchedulerException e) {
            log.error("删除任务失败", e);
        }
        return  false;
    }

可以看到Quartz创建任务就是通过JobDetail和Trigger来实现定时任务的,JobDetail声明了任务的一些信息,比如名称、分组、描述、需要执行类(例如:com.chuanz.task.CheckExecutor)需要的数据(通过调用setJobData方法) Trigger设置了任务的执行周期及触发时间,这里通过cron表达式来设置触发,另外Trigger还有另外一种触发方式,通过SimpleScheduleBuilder来实现,比如:

.withSchedule(SimpleScheduleBuilder.repeatMinutelyForTotalCount(10, 5))

我需要重复执行10次,每隔5分钟执行一次

创建我们需要执行的任务,实现Job接口,并重写execute方法

@Slf4j
@Service
public class CheckExecutor implements Job {

    @Override
    public void execute(JobExecutionContext context) throws JobExecutionException {
        log.info("执行计划冲突类型审核开始");
        // 获取数据
        ScheduleTaskConfig config = (ScheduleTaskConfig)context.get("config");

        /**
         * 处理逻辑
         */
        log.info("执行计划冲突类型审核结束");
    }
}

我们可以将quartz创建、删除、停止、恢复任务都封装成一个个接口,通过前端页面来管理quartz的任务,具体前端页面代码我就不贴了,基本就是调用QuartzService里面的方法了,另外说说获取quartz任务列表的时候,我们可以获取Trigger里面响应的执行状态,通过Trigger里面的方法来获取,比如:

JobKey jobKey = new JobKey(config.getName(), config.getGroupName());
List<Trigger> triggers = (List<Trigger>) scheduler.getTriggersOfJob(jobKey);
if (triggers.size() > 0) {
    Trigger trigger = triggers.get(0);
    //开始执行时间
    trigger.getStartTime();
    //上一次执行时间
    trigger.getPreviousFireTime();
    //下一次执行时间
    trigger.getNextFireTime();
}

trigger有很多实现,比如我们这里使用的是cron表达式,所以相对应的就是CronTriggerImpl,常用的触发器有四种。

触发器介绍

SimpleTrigger:简单的触发器

指定从某一个时间开始,以一定的时间间隔,如秒、分、小时重复执行的任务
比如:从10:00 开始,每隔1小时/1分钟/1秒钟执行一次,执行10次。
对应的方法如下:

8A76CC1568BDC509456138BBB5327AB8

CalendarIntervalTrigger:日历触发器

类似于SimpleTrigger,指定从某一个时间开始,以一定的时间间隔执行的任务。 但是不同的是SimpleTrigger指定的时间间隔为毫秒,没办法指定每隔一个月执行一次(每月的时间间隔不是固定值),而CalendarIntervalTrigger支持的间隔单位有秒,分钟,小时,天,月,年,星期。
对应的方法如下:

0D2A2B1FFAA7D216938E00C9D6E3DAAC

DailyTimeIntervalTrigger:日期触发器

指定每天的某个时间段内,以一定的时间间隔执行任务。并且它可以支持指定星期。 它适合的任务类似于:指定每天9:00 至 18:00 ,每隔70秒执行一次,并且只要周一至周五执行。 比如如下代码:

DailyTimeIntervalTrigger trigger = dailyTimeIntervalSchedule()
    .startingDailyAt(TimeOfDay.hourAndMinuteOfDay(10, 0)) //从10:00点开始
    .endingDailyAt(TimeOfDay.hourAndMinuteOfDay(22, 0)) //22:00点结束
    .onDaysOfTheWeek(MONDAY,TUESDAY,WEDNESDAY,THURSDAY,FRIDAY) //周一到周五执行
    .withIntervalInHours(1) //每间隔1小时执行一次
    .withRepeatCount(50) //最多重复50次(实际执行50+1次)
    .build();

CronTrigger:Cron表达式触发器

适合于更复杂的任务,它支持类型于Linux Cron的语法(并且更强大)。基本上它覆盖了以上三个Trigger的绝大部分能力
属性只有一个就是cron表达式

我们可以根据对应的触发器取到任务执行的数据,如之前用SimpleScheduleBuilder.repeatMinutelyForTotalCount(10, 5),那么对应的就是SimpleTrigger,我们可以获取执行的次数,剩余的次数等。 触发器的介绍可以参考官方文档

 

 

 

Redis导致接口变慢故障排查

Redis导致接口变慢故障排查

近段时间我们一个接口总是隔断时间出现一次访问很慢的情况,如下图,这是我们通过kibana统计的接口响应时间

WechatIMG12-1

最开始我们想到是不是并发量太大导致后端数据库压力太大了,所以多开了一个实例,并且数据读写采用了读写分离,但情况依旧,最后也打印出了操作数据库、Redis的耗时情况,如下图

WechatIMG13

发现MySQL并不是瓶颈,Redis读取的时候耗时比MySQL更严重,我们知道Redis是单线程直接操作内存的,一定是有某些操作阻碍了主线程的执行,查看了Redis执行日志

WechatIMG14

发现在耗时那个点,Redis正在做持久化操作,而且使用的是RDB全量快照的方式,介绍下RDB的持久化功能:
Redis默认是使用RDB的持久化策略,可以配置周期性将数据保存到磁盘,比如可配置在1分钟内发生1000次写操作,就保存一次,这里的保存是全量保存,即Redis会fork一个子进程来循环所有的数据,然后将数据写入到RDB文件中,如果在某个时间段有频繁的写请求过来,那么Redis就不不断的fork子进程来处理数据库快照操作,但fork操作会发生堵塞,所以那段时间就会发生客户端的读写请求比较卡的情况,Redis的持久化策略流程如下图:

3084708676-5b70e0fd04072_articlex

解决,使用AOP持久化策略或者我们可以配置不使用Redis持久化策略,因为根据接口的业务情况,发现即使数据丢失,也不会造成太大影响,可以直接再去读数据库获取,具体配置只要在最后一行加上:save “”,即可禁用RDB

参考:

http://www.cnblogs.com/zhoujinyi/archive/2013/05/26/3098508.html

https://segmentfault.com/a/1190000015983518

 

 

Redis分布式锁使用

最近项目中遇到同时有两个线程同时更新一行记录导致后面一条语句执行失败的问题,由于项目是部署在不同的服务器上,这里要控制两个线程的执行顺序,自然想到了使用Redis的锁,废话不多说,下面给出具体实现

/**
 * 核查四要素相同报文是否正在处理,如果有实例正在处理四要素相同报文pass,否则线程等待
 * 
 * @param processData
 */
public void checkPacketProcessRepeat(ProcessData processData) {
	try {
		// 四要素key
		String repeatKey = REPEATKEYSTART + processData.getReviseFlight().getKey();
		while (true) {
			// 设置nx锁,如果nx锁设置成功跳出去,继续执行报文后续处理流程
			if (setNX(repeatKey, EXIST, 3)) {
				log.info("FlightPreProcess-checkPacketProcessRepeat,报文处理拿到NX锁,直接pass,key:{},sourceId:{}", processData.getReviseFlight().getKey(), processData.getReviseFlight()
						.getSourceId());
				return;
			}
			// 如果有当前航班有nx锁、或者nx锁设置失败,则需要等待3秒,等待其他实例处理完成
			log.info("FlightPreProcess-checkPacketProcessRepeat,报文多实例并发处理,需要等待3秒,key:{},sourceId:{}", processData.getReviseFlight().getKey(), processData.getReviseFlight()
					.getSourceId());
			Thread.sleep(3000);
		}
	} catch (Exception e) {
		log.error("FlightPreProcess-checkPacketProcessRepeat,异常,key:{},e:{}", processData.getReviseFlight().getKey(), e);
	}
}

这里根据报文的四要素确定唯一条记录,先调用setNX获取redis锁,如果获取到了就执行后面的逻辑,如果没有获取到则等待3s再重试,下面是setNX方法的实现

/** 设置锁 */
private boolean setNX(final String key, String value, final int exp) {
	return (Boolean) redisTemplate.execute(new RedisCallback<Object>() {
		@Override
		public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
			byte[] serializeKey = redisTemplate.getStringSerializer().serialize(key);
			Boolean acquire = connection.incr(serializeKey) == 1;
			// 如果设值成功,则设置过期时间
			if (acquire) {
				connection.expire(serializeKey, exp);
			}
			return acquire;
		}
	});
}

这里使用了redis的incr命令,它是一个原子操作,如果key不存在,那么key的值将初始化为0,然后执行INCR操作,这里判断如果设置成功,则对key设置过期时间,相当于了一个带有时间的锁。

在Redis2.6.12版本后,使用set命令也可以实现分布式锁,具体代码如下:

public static Boolean setNX(final String key, final String value, final int exp) {
        return (Boolean) redisTemplate.execute(new RedisCallback<Object>() {
			@Override
			public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
				Jedis jedis = (Jedis) connection.getNativeConnection();
				String result = jedis.set(key, value, "NX", "EX", exp);
				if ("OK".equals(result)) {
					return Boolean.TRUE;
				}
				return Boolean.FALSE;
			}
		});
    }

这里重点说说第3个和第4个参数,这里填的是NX,意思是当key不存在时,我们进行set操作,若key已经存在则不进行任何操作,第4个表示我们要给key设置一个过期时间,具体时间由第5个参数决定。

另外我们这里保存了key对应的value值,所以线程可以根据value值来释放锁,这里的value值可以是线程的ID,比如我们线程后面的逻辑执行失败了,我们可以通过这个value值来尽快释放锁,减少其它线程的等待时间,我们可以使用Lua脚本来实现

private static final Long RELEASE_SUCCESS = 1L;
private static final RELEASE_LOCK_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";

public static Boolean releaseLock(final String key, final String value) {
			return (Boolean) redisTemplate.execute(new RedisCallback<Object>() {
			@Override
			public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
				Jedis jedis = (Jedis) connection.getNativeConnection();
				Object result = jedis.eval(RELEASE_LOCK_SCRIPT, Collections.singletonList(key),
						Collections.singletonList(value));
				if (RELEASE_SUCCESS.equals(result)) {
					return Boolean.TRUE;
				}
				return Boolean.FALSE;
			}
		});
    }

通过Lua脚本获取对应key的value值,如果value值和给定的一样,则释放锁

 

Spring Boot启动出现死锁

Spring Boot启动出现死锁

最近公司的一个项目在启动后,在调用往mongoDB写数据时,一直写不进去,运行一段时间后,程序出现内存溢出,jstack导出了线程信息,如下:

CC6B8F5EF4392AAA4F0B393FFDA32AA6 DFB379909A737E20E4A894BFF683739A

mongoDB在插入数据时发生了堵塞,在等待一个监控对象,主线程在调用afterPropertiesSet方法,一直在等待中,表示spring的初始化工作一直没有完成,所以我们结合代码分析,查看afterPropertiesSet方法

    @Override
    public void afterPropertiesSet() throws Exception {
        scheduleRulePriorityCacheService.start();

        if (isTesting == false) {
            kafkaConsumerTask.start();
            executeService.execute();
        }

    }

 

在spring属性设置完成后,执行了一个kafka消费任务,这个kafka消费任务大致的流程就是从kafka中取出数据,然后放入一个队列中,之后执行了executeService.execute()从队列里面拿出数据然后进行业务处理,我们看看execute()方法

	public void execute() {
		while (true) {
			try {
				if (receiveQueue.isEmpty()) {
					Thread.sleep(200);
				} else {
					String msg = receiveQueue.poll();
					if (StringUtils.isBlank(msg)) {
						continue;
					}
					poolTaskExecutor.execute(() -> {
						scheduleProcessCenter.process(msg);
					});
				}
			} catch (Exception e) {
				logger.error("ExecuteService error,e:{}", e);
			}
		}
	}

这里就是一个无线循环操作,即如果队列有数据就消费,没有就等待200ms

我们知道,afterPropertiesSet方法是spring Bean的生命周期的一部分,这里发生了死锁,必定是锁住了一个资源,没有释放,而MongoDB又需要这个资源,我看到错误信息,在调用DefaultSingletonBeanRegistry.getSingleton方法时锁住了一个资源,我们查看spring的源代码,

7D5A37364ED4463F83CA98076DBEB6AD

可以看到锁住了singletonObjects对象,这个对象就是spring的单例容器,同样我们可以确定,在MongoDB调用注册事件的时候也需要这个对象,我们同样查看源代码

CECB8DBCEB5F05A176976D877FCEC3A1

可以看到MongoDB在注册事件的时候同时需要锁住retrievalMutex对象,那么retrievalMutex和singletonObjects有什么关系?我们接着看

ADE46D9008DA9E92021028F3E5582578 1F3CF43276456CBF7BF938AE5A1CC168

点进去查看getSingletonMutex方法,返回的就是singletonObjects对象,所有retrievalMutex实质和singletonObjects是同一个对象

那么就很容易得出结论了,在调用afterPropertiesSet方法时,singletonObjects对象一直没释放,而MongoDB又需要这个对象,所以产生了死锁。 解决方法,让afterPropertiesSet方法尽快释放singletonObjects对象,我们可以开启一个新线程来执行从队列中读取数据做业务处理的逻辑

	public void execute() {
		poolTaskExecutor.execute(this);
	}

	@Override
	public void run() {
		while (true) {
			try {
				if (receiveQueue.isEmpty()) {
					Thread.sleep(200);
				} else {
					String msg = receiveQueue.poll();
					if (StringUtils.isBlank(msg)) {
						continue;
					}
					poolTaskExecutor.execute(() -> {
						scheduleProcessCenter.process(msg);
					});
				}
			} catch (Exception e) {
				logger.error("ExecuteService error,e:{}", e);
			}
		}
	}

 

Mysql死锁的问题

Mysql死锁的问题

1,先看错误日志

WechatIMG21 WechatIMG22

从以上日志可以看出是两条SQL执行出现了问题,后面一条SQL回滚了

SQL为:

update dynamic_check_packet_system set check_flag = '3' where create_time <= '2018-12-19 02:00:00' and check_flag='0' and conflict_field not in ('suspectCancel');
update flight.dynamic_check_packet_system set check_flag='1', update_time='2018-12-19 09:55:00.0',check_start_time='2018-12-19 10:00:00' where id=14211034;

使用explain查看第一条SQL

E9C461C524DED9A648DC4DCFF089919F

可以看到,这个语句使用了idxcheckflag这个索引,createtime在前,为什么没有使用createtime?

查看表中<=createtime的数据,发现有14166149条,远远大于checkflag=0的记录数,这时MySQL就会优先选择使用chekflag的索引,所以第一条语句会把checkflag=0的所有记录数都锁住。

第二条SQL同样更新check_flag=0,id=14211034的记录,但这条记录是被第一条SQL锁住的,所以就会更新失败了?

深入分析:
这里查看MySQL引擎,用的是Innodb,Innodb是支持行锁的,既然第一条SQL把这条记录行锁住了,第二条SQL应该等待才对,为什么会发生死锁呢?所以这里一定存在两把锁,而且锁的顺序不同。

我们知道MySQL的Innodb主键使用了聚集索引(索引直接指向实际数据),而如果再新建一个索引,这个索引会指向主键索引,然后通过主键索引找到数据,所以这里存在需要更新聚集索引ID数据和二级索引check_flag数据

第一条语句通过checkflag=0查找,那么就先会锁住二级索引checkflag数据,然后再去获取聚集索引ID数据的锁

而第二天SQL则是通过ID查找,那么就会先会锁住聚集索引ID数据,然后再去获取二级索引check_flag数据

显然这样获取锁的先后顺序不同,就造成了死锁。

问题解决:
让第一条SQL语句使用createtime索引,可以指定一个createtime>’开始时间’ and createtime<’结束时间’来减少扫描行数,让MySQL优先使用createtime索引。