MySQL事务隔离级别

Posted on by

对于数据库的隔离级别之前一直没有做详细整理，最近项目运行中发现了一个问题，所以抽时间对这块认真研究了下

业务场景：
服务A在处理流程中，会调用外部服务B，然后写入一条数据，服务B执行完成后，会回调服务C的接口更新服务A写入的数据。
问题：
在服务B回调服务C的时候总是找不到服务A写入的数据，在服务C中添加延时重试，问题依然存在，但此时查看数据库，对应的数据是已经存在。

先说原因吧，是因为MySQL的事务默认隔离级别是：可重复读。
在服务A调用服务B后，还没有写入数据到数据库，服务B就已经回调服务C了，服务C此时肯定是找不到对应的数据的，由于MySQL默认隔离级别是可重复读（即在一个事务中，对于同一份数据读取都是一样的），所以即使服务A已经写入了数据，服务C依然读取不到。

解决方案：
在服务C中的查询，不要放到一个事务里面，单独提取一个方法，后面的更新逻辑放到同一个事务中

什么是事务？

数据库事务是数据库管理系统执行过程中的一个逻辑单位，由一个有限的数据库操作序列构成。
比如：某人在商店购买100商品，其中包括两个操作：
1.该人账户减少100元
2.商店账户增加100元
这两个操作要么同时执行成功，要么同时执行失败。

数据库事务的ACID性质

  • Atomic：原子性，将所有SQL作为原子工作单元执行，要么全部执行，要么全部不执行;
  • Consistent：一致性，事务完成后，所有数据的状态都是一致的，即该人的账户减少了100，商店的账户必须增加100
  • Isolation：隔离性，比如两个人从同一个账户取款，这两个事务对数据的修改必须相互隔离，具体隔离策略后面具体讲解。
  • Duration：持久性，事务完成后，对数据库数据的修改必须被持久化存储。

数据库的隔离级别

在单个事务中，不需要做隔离，所谓数据库隔离级别是针对在并发事务的情况下，解决导致的一系列问题，这些问题包括：脏读、不可重复读、幻读，具体隔离级别如下图：

隔离级别 脏读 不可重复读 幻读
SERIALIZABLE（串行化） 避免 避免 避免
REPEATABLE READ（可重复读） 避免 避免 允许
READ COMMITED（读已提交） 避免 允许 允许
READ UNCOMMITED（读未提交） 允许 允许 允许

SERIALIZABLE（串行化）

当两个事务同时操作数据库中相同数据时，如果第一个事务已经在访问该数据，第二个事务只能停下来等待，必须等到第一个事务结束后才能恢复运行。因此这两个事务实际上是串行化方式运行。

REPEATABLE READ（可重复读）

一个事务在执行过程中可以看到其他事务已经提交的新插入的记录，但是不能看到其他事务对已有记录的更新。

READ COMMITTED（读已提交数据）

一个事务在执行过程中可以看到其他事务已经提交的新插入的记录，而且还能看到其他事务已经提交的对已有记录的更新。

READ UNCOMMITTED（读未提交数据）

一个事务在执行过程中可以看到其他事务没有提交的新插入的记录，而且还能看到其他事务没有提交的对已有记录的更新。

没有事务隔离级别导致的问题

如果没有数据库的隔离级别，数据库的数据是实时变化的，即每个事务都可以读到其它事务修改后的数据，下面结合实例介绍每个场景的问题。

脏读

定义：读到未提交更新的数据

时间点 事务1 事务2
T1 开始事务
T2 开始事务
T3 查询账户余额为1000
T4 取出100后金额为900
T5 查询账户金额为900（脏读）
T6 撤销事务，余额恢复为1000
T7 存入100元后，金额变为1000
T8 提交事务

如上事务1取出金额100后又回滚了，即啥都没做，但事务2存入了100，但最终的金额确还是1000，正确应该是1100。
在T5时间节点出现了脏读，如果数据库配置了隔离级别为SERIALIZABLE、REPEATABLE READ、READ COMMITTED，在事务1没有提交的时候，事务2读取的都是原来的值就不会出现问题。

不可重复读

定义：在同一个数据中，两次读取到的数据不一致，读到了其他数据提交更新的数据

时间点 事务1 事务2
T1 开始事务
T2 开始事务
T3 查询账户余额为1000
T4 查询账户余额为1000
T5 取出100后金额为900
T6 提交事务
T7 查询账户余额为900(与T4读取的不一致)

事务2的两次读取到的数据不一致，第二次读取到了事务1提交的数据

幻读

定义：读取到另一个事务已提交插入或删除的数据。

时间点 事务1 事务2
T1 开始事务
T2 开始事务
T3 统计一年级1班所有的学生人数为40人
T4 一年级1班新增一名学生
T5 提交事务
T6 再次统计一年级1班的所有学生人数为41人

事务2第一次统计人数为40人，第二次统计为41人，两次统计的结果不一致，同样如果T4时间节点转走一名学生，也会出现不一致

不可重复读和幻读看起来比较类似，都是一个事务里面读取到两次不同的结果
本质的区别是：不可重复读是由于数据更新导致数据不一致导致，幻读是由于插入或删除了数据导致的。

ElasticSearch高并发场景写入优化

Posted on by

ElasticSearch高并发场景写入优化

ElasticSearch号称分布式全文搜索引擎，但使用不当依然会很慢，特别是在高并发写入时，会存在写入超时的问题。

在公司内部，基本所有的日志都会放入到ElasticSearch，比如接口访问时间日志、动态/计划数据审核日志、动态/计划抓取报文日志、动态报文更新日志等，每天的写入数据量巨大，最初我们基于ElasticSearch封装了一层，使用内部RestHighLevelClient获取数据，部分代码如下：

@Slf4j
public class ElasticsearchService {

    private RestHighLevelClient client;

    public ElasticsearchService(RestHighLevelClient client) {
        if (client == null)
            throw new IllegalArgumentException("client");
        this.client = client;
    }

    /**
     * 根据id获取文档
     **/
    public GetResponse getDocument(String index, String type, String id) {
        if (StringUtils.isAnyBlank(index, type, id))
            return null;
        try {
            GetRequest getRequest = new GetRequest(index, type, id);
            GetResponse getResponse = client.get(getRequest, RequestOptions.DEFAULT);

            return getResponse;

        } catch (Exception e) {
            log.error("获取文档失败", e);
        }
        return null;
    }

    /**
     * 插入文档， id可以为空，id如果为空，就自动生成id
     **/
    public IndexResponse insertDocument(String index, String type, String id, String json) {
        if (StringUtils.isAnyBlank(index, type, json))
            return null;

        try {
            IndexRequest indexRequest = generateInsertRequest(index, type, id, json);

            IndexResponse indexResponse = client.index(indexRequest, RequestOptions.DEFAULT);
            return indexResponse;
        } catch (Exception e) {
            log.error("创建文档失败", e);
        }
        return null;
    }
}

然而在放到线上生成环境下，经常会出现写入失败数据丢失的情况，发现是数据量写入太大，调用接口超时，直接返回错误了。RestHighLevelClient实质就是通过httpclient请求接口发送数据，是基于http协议。

查看Elasticsearch文档，发现Elasticsearch同样支持使用Transport传输，其内部使用netty长连接通信，并且可以设置处理连接的数量。

新建一个TransportClientFactory用来创建TransportClient，这里连接需要使用密钥，具体填写自己Elasticsearch的账号密码。

public class TransportClientFactory {
    // 默认处理核心数,cpu核心数
    private static int DEFAULT_PROCESSORS = 8;
    // 配置核心数
    private static int processors = 8;

    public static TransportClient create(String addresses, int port, String keyStore, String keyPassword, String trustStore, String trustPassword) throws Exception {

        if (StringUtils.isBlank(addresses)) {
            throw new IllegalArgumentException("请输入ES的地址，多个地址之间用逗号分隔");
        }

        if (StringUtils.isAnyBlank(keyStore, keyPassword, trustStore, trustPassword)) {
            throw new IllegalArgumentException("缺少证书或密码");
        }

        String[] addrs = addresses.split(",");
        if (addrs.length == 0) {
            throw new IllegalArgumentException("请输入ES的地址，多个地址之间用逗号分隔");
        }

        List<String> validAddrs = new ArrayList<>();
        for (String addr : addrs) {
            if (StringUtils.isNotBlank(addr)) {
                validAddrs.add(addr.trim());
            }
        }

        if (validAddrs.size() == 0) {
            throw new IllegalArgumentException("请输入ES的地址，多个地址之间用逗号分隔");
        }

        if (processors < DEFAULT_PROCESSORS) {
            processors = DEFAULT_PROCESSORS;
        }
        if (processors > 32) {
            processors = 32;
        }
        int threadPoolCore = processors * 2;
        int threadPoolMax = processors * 2;
        Settings settings = Settings
                .builder()
                .put("path.home", ".")
                // .put("cluster.name", clusterName)
                .put("client.transport.ignore_cluster_name", true).put("searchguard.ssl.transport.enabled", true)
                .put("searchguard.ssl.transport.enforce_hostname_verification", false).put("searchguard.ssl.transport.keystore_filepath", keyStore)
                .put("searchguard.ssl.transport.keystore_password", keyPassword).put("searchguard.ssl.transport.truststore_filepath", trustStore)
                .put("searchguard.ssl.transport.truststore_password", trustPassword).put("processors", processors).put("thread_pool.flush.core", threadPoolCore)
                .put("thread_pool.flush.max", threadPoolMax).build();

        TransportClient client = new PreBuiltTransportClient(settings, SearchGuardPlugin.class);

        for (String validAddr : validAddrs) {
            client.addTransportAddress(new TransportAddress(InetAddress.getByName(validAddr), port));
        }

        return client;
    }

    public static int getProcessors() {
        return processors;
    }

    public static void setProcessors(int processors) {
        TransportClientFactory.processors = processors;
    }

}

修改ElasticsearchService

@Slf4j
public class ElasticsearchService {

    /**
     * 默认10秒超时
     **/
    private long bulkTimeoutSecond = 10;

    private TransportClient client;

    public ElasticsearchService(TransportClient client) {
        this.client = client;
    }

    public ElasticsearchService(TransportClient client, int asyncTimeoutSecond) {
        this.client = client;
        this.bulkTimeoutSecond = asyncTimeoutSecond;
    }

    public GetResponse getDocument(String index, String type, String id) {
        if (StringUtils.isAnyBlank(index, type, id))
            return null;

        return this.client.prepareGet(index, type, id).get();
    }

    public IndexResponse createDocument(String index, String type, String id, String msg) {
        if (StringUtils.isAnyBlank(index, type, msg))
            return null;

        IndexRequestBuilder builder = createDocumentRequestBuilder(index, type, id, msg);
        if (builder == null)
            return null;

        return builder.get();
    }

    private IndexRequestBuilder createDocumentRequestBuilder(String index, String type, String id, String msg) {
        if (StringUtils.isAnyBlank(index, type, msg))
            return null;

        if (StringUtils.isBlank(id)) {
            return this.client.prepareIndex(index, type).setSource(msg, XContentType.JSON);
        } else {
            return this.client.prepareIndex(index, type, id).setSource(msg, XContentType.JSON);
        }
    }

}

另外为了提升写入效率，可以批量异步一次性写入，使用bulk方法，如下：

  /**
     * 批量操作，listener可空，支持 IndexRequest， UpdateRequest， DeleteRequest
     **/
    public void bulkDocumentOperationAsync(List<DocWriteRequest<?>> requests, ActionListener<BulkResponse> listener) {
        if (requests == null || requests.size() == 0)
            return;

        BulkRequest request = prepareBulkRequests(requests);
        if (request == null)
            return;

        this.client.bulk(request, listener);
    }

    /**
     * 支持 IndexRequest， UpdateRequest， DeleteRequest
     **/
    private BulkRequest prepareBulkRequests(List<DocWriteRequest<?>> requests) {
        BulkRequestBuilder builder = bulkRequestBuilder();

        for (DocWriteRequest<?> request : requests) {
            if (request instanceof IndexRequest) {
                builder.add((IndexRequest) request);
            } else if (request instanceof UpdateRequest) {
                builder.add((UpdateRequest) request);
            } else if (request instanceof DeleteRequest) {
                builder.add((DeleteRequest) request);
            }
        }

        if (builder.numberOfActions() == 0) {
            return null;
        }
        return builder.request();
    }

在dao层，我们可以一次性写入多条数据

    /**
     * 异步写入动态日志
     * 
     * @param flightRecords
     */
    public void insertFlightRecodeListAsync(List<FlightRecord> flightRecords) {
        try {
            if (CollectionUtils.isEmpty(flightRecords)) {
                return;
            }
            ElasticsearchService elasticsearchService = elasticsearchProvider.getElasticsearchService();
            List<DocWriteRequest<?>> requests = new ArrayList(flightRecords.size());
            for (FlightRecord flightRecord : flightRecords) {
                // index以航班日期为准
                String index = elasticsearchProvider.FLIGHT_RECORD_INDEX_PREFIX + flightRecord.getLocalDate().replace("-", ".");
                requests.add(elasticsearchService.createDocumentRequest(index, elasticsearchProvider.FLIGHT_RECORD_TYPE, null, JSON.toJSONString(flightRecord)));

            }
            if (CollectionUtils.isEmpty(requests)) {
                return;
            }
            elasticsearchService.bulkDocumentOperationAsync(requests, new ActionListener<BulkResponse>() {
                @Override
                public void onResponse(BulkResponse bulkItemResponses) {
                }

                @Override
                public void onFailure(Exception e) {
                    log.error("flightRecord写入ES失败", e);
                }
            });
        } catch (Exception e) {
            log.error("insertFlightRecodeListAsync Exception", e);
        }
    }

其它写入性能优化：

  1. 去掉不必要的字段分词和索引，ElasticSearch会默认对所有字段进行分词，一般查询我们都是根据航班号、航班日期、起飞机场、到达机场查询；所以没必要的字段，我们可以使用keyword类型，不进行分词。 
    PUT my_index
{
  "mappings": {
    "my_type": {
      "properties": {
        "tail_no": { 
          "type": "keyword",
          "index": false
        }
      }
    }
  }
}
  2. 对于一些普通的日志，比如动态/计划抓取网页日志，数据丢失也无所谓，可以禁用掉refresh和replia，即index.refreshinterval设置为-1，将index.numberof_replicas设置为0即可。 
    PUT my_index
{
  "settings": {
    "index": {
        "refresh_interval" : "-1",
        "number_of_replicas" : 0
    }
  }
}
  3. 使用ElasticSearch自增ID，如果我们要手动给ElasticSearch document设置一个id，那么ElasticSearch需要每次都去确认一下那个id是否存在，这个过程是比较耗费时间的。如果我们使用自动生成的id，那么ElasticSearch就可以跳过这个步骤，写入性能会更好。
  4. 使用多线程写入ElasticSearch，单线程发送bulk请求是无法最大化ElasticSearch集群写入的吞吐量的。如果要利用集群的所有资源，就需要使用多线程并发将数据bulk写入集群中。为了更好的利用集群的资源，这样多线程并发写入，可以减少每次底层磁盘fsync的次数和开销。一样，可以对单个ElasticSearch节点的单个shard做压测，比如说，先是2个线程，然后是4个线程，然后是8个线程，16个，每次线程数量倍增。一旦发现es返回了TOOMANYREQUESTS的错误，JavaClient也就是EsRejectedExecutionException，此时那么就说明ElasticSearch是说已经到了一个并发写入的最大瓶颈了，此时我们就知道最多只能支撑这么高的并发写入了。
  5. 适当增加index buffer的大小，如果我们要进行非常重的高并发写入操作，那么最好将index buffer调大一些，indices.memory.indexbuffersize，这个可以调节大一些，设置的这个index buffer大小，是所有的shard公用的，但是如果除以shard数量以后，算出来平均每个shard可以使用的内存大小，一般建议，但是对于每个shard来说，最多给512mb，因为再大性能就没什么提升了。ElasticSearch会将这个设置作为每个shard共享的index buffer，那些特别活跃的shard会更多的使用这个buffer。默认这个参数的值是10%，也就是jvm heap的10%，如果我们给jvm heap分配10gb内存，那么这个index buffer就有1gb，对于两个shard共享来说，是足够的了。

Kafka分区分配策略

Posted on by

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实现分表

Posted on by

使用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排序（三）

Posted on by

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创建索引及数据（二）

Posted on by

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安装及介绍（一）

Posted on by

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

Posted on by

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导致接口变慢故障排查

Posted on by

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分布式锁使用

Posted on by

最近项目中遇到同时有两个线程同时更新一行记录导致后面一条语句执行失败的问题，由于项目是部署在不同的服务器上，这里要控制两个线程的执行顺序，自然想到了使用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值和给定的一样，则释放锁