Thinkphp + go-mysql-elasticsearc + elasticsearch

发表于 2022-01-12

前言

项目使用 thinkphp + mysql 对列表进行查询，列表数据量达到千万级别，且搜索维度效果，故采用elasticsearch进行搜索

为了尽量不改动代码逻辑，需要实现

数据自动同步至ES ( 使用 go-mysql-elasticsearch）
平滑切换TP自带的数据库查询操作到ES（自定义 query，转化TP 的查询条件为 ES的查询条件）

graph TD
A --> | go-mysql-elasticsearc 初始化时自动导出数据为SQL|G(sql)
A[mysql] --> |mysql开启log row格式|B(binlog)

G --> |解析SQL文件|D
B --> |伪装成从库  请求 数据| D[go-mysql-elasticsearch]
D -->|同步数据|F(elasticsearch)

Z[ES 数据流程图]

功能使用

说明

php 中\think\db\query\Elasticsearch 为处理 TP语法到 ES 中的的查询语法，调用方式通DB 相同

暂只支持where , whereor ,wherein , limit , order ,find select ,value ,sum 方法

查询方式

1	Elasticsearch\Elasticsearch\FinanceBillIncome::where('xxx','xxx')->find();

列表时先通过ES 查询出主键ID, 再通过主键查询出明细数据
列表查询通过 count 方法
列表头部统计，通过如下方式查询

->sum(   [
                     'expect_fee',
                     'adjustment_fee',
                     'confirm_fee',
                 ]);

暂不支持 join，必须join 的降级为DB 处理
alias 方法指定后，ES 查询时将自动去除查询条件与字段中头部别名

使用步骤

先通过

.env

# 搜索引擎地址
ELASTICSEARCH_HOST_NAME='192.168.x'x.133'
ELASTICSEARCH_HOST_PORT='9200'
ELASTICSEARCH_INDEX_PREFIX='xxx_'

database.php

'elasticsearch'=>[
    'prefix' =>  env('ELASTICSEARCH_INDEX_PREFIX'),
    'type'     => 'elasticsearch',
    'hostname' => env('ELASTICSEARCH_HOST_NAME'),
    'hostport' => env('ELASTICSEARCH_HOST_PORT'),
    'debug' => true,
    // 是否严格检查字段是否存在
    'fields_strict' => false,
    // 数据集返回类型 array 数组 collection Collection对象
    'resultset_type' => 'array',
    'sql_explain' => false,
    'query'           => '\\think\\db\\query\\Elasticsearch',
],

builder 文件
connector文件
query文件
新增 class 如

1	Elasticsearch\Elasticsearch\FinanceBillIncome

初始化 ES，创建 index

1	/cron/elasticsearch/initSearch

输出的内容为 go-mysql-elasticsearch 中的配置
启动 go-mysql-elasticsearch

监控

定时 monitor 表中更新时间
定时查询ES 中monitor 的时间
对比时间差异，差异时间为延迟时间

php-resque延迟执行与失败重试

发表于 2018-02-22

一、前言

使用第三方库进行短信发送等，为了响应速度，失败延迟重试等，决定使用消息队列。

二、使用

php-resque 在网上有不少教程在此便不加以累赘，参考地址: PHP-resque使用经验总结

三、结构分析

从push与pop方法中可看出，队列中使用redis的List（列表）右侧插入数据，左侧取出数据。


class Resque(){



    	public static function push($queue, $item)
	{
		self::redis()->sadd('queues', $queue);
		self::redis()->rpush('queue:' . $queue, json_encode($item));
	}
	
		public static function pop($queue)
	{
		$item = self::redis()->lpop('queue:' . $queue);
		if(!$item) {
			return;
		}

		return json_decode($item, true);
	}
	
}

四、结构设计

List结构不能完成延迟执行这一功能，为此新增两个redis key 分别为SortedSet（有序集合）与Hash（哈希表），
将队列数据放在Hash表中，将数据的Key 放入 SortedSet 中其中到达可执行的时间戳为Sort ，当 Sort 小于当前时间时，则将Hash 表中的数据移动至 List 队列中右侧，排队执行

五、代码实现

Resque_Job 中

$this->instance->queue = $this->queue;

后插入

1 2	//执行次数 $this->instance->try_time = isset($this->payload['try_time'] ) ? $this->payload['try_time'] : 1;

Resque_Job 中插入3个方法



class Resque_Job
{
    /**
     * 超时执行
     * @param      $queue
     * @param      $class
     * @param null $args
     * @param      $delay
     * @param bool $monitor
     *
     * @return string
     */
	public static  function delay($queue, $class, $args = null, $delay=60,$monitor = false){
        if($args !== null && !is_array($args)) {
            throw new InvalidArgumentException(
                'Supplied $args must be an array.'
            );
        }
        $id = md5(uniqid('', true));

        Resque::later($queue,
            array(
                'class'	=>$class,
                'args'	=> array($args),
                'id'	=>$id,
            ),$delay);



        if($monitor) {
            Resque_Job_Status::create($id);
        }
        return $id;

	}

    /**
     * 重试
     * @param $delay
     */
    public function delayJob($delay)
    {
        $this->payload['try_time'] =( isset($this->payload['try_time'])&& $this->payload['try_time']>1) ? $this->payload['try_time'] : 1;
        $this->payload['try_time']++;

        return Resque::later($this->queue, $this->payload, $delay);
    }

    /**
     *  加载延迟数据
     *  到达延迟时间的 Job 从Hash移动至 List队列
     * @param $queue
     *
     * @return bool
     */
	public static function loadDelay($queue){
               return  Resque::move($queue);
	}
}

reidis 支持

class Resque_Redis extends Redisent {
    private $keyCommands = array(
            ...
    	'zremrangebyscore',
	'sort',
	//插入这3个类型
        'hset',
        'hget',
        'hdel'
    )
}

class Resque_RedisCluster extends RedisentCluster{
    private $keyCommands = array(
            ...
    	'zremrangebyscore',
	'sort',
	//插入这3个类型
        'hset',
        'hget',
        'hdel'
    )
}

Resque_Worker 每次判断是否有到达时间的Job

class Resque_Worker{
	public function work($interval = 5)
	{
		$this->updateProcLine('Starting');
		$this->startup();
		while(true) {
			if($this->shutdown) {
				break;
			}
			$this->loadDelay();// 加载延迟执行
			
			... 
			}
     }
    
      /**
     * 延迟执行 移动队列位置
     */
	public function loadDelay(){
        $queues = $this->queues();
        if($queues){
        	foreach ($queues as $queue){
                $this->log('Checking Delay' . $queue, self::LOG_VERBOSE);
                 Resque_Job::loadDelay($queue);

	        }
        }

	}
}

class Resque 中加入3个方法

class Resque{
    
    public static function later($queue,$item,$delay){
        self::redis()->sadd('queues', $queue);
        self::redis()->hset('delays:'.$queue, $item['id'], json_encode($item));
        self::redis()->zadd('delay:'.$queue,time() + $delay, $item['id']);
    }

    /**
     * 从 哈希数据移动到队列
     *
     * @param $queue
     */
    public static function move($queue){
     // 获取前一千个满足条件的数据
      $zDelayKeys =   self::redis()->zrangebyscore('delay:'.$queue,0,time(),'limit',0,1000);

      if($zDelayKeys ){
          foreach ($zDelayKeys as $delayKey){
              $item = self::redis()->hget('delays:'.$queue,$delayKey);
              if ($item) {
                  $delete = self::redis()->hdel('delays:' . $queue, $delayKey);
                  if ($delete > 0) { //可能有多个WORK 以删除为准


                      self::redis()->sadd('queues', $queue);
                      self::redis()->rpush('queue:' . $queue, $item); //无需再json_encode



                      self::redis()->zrem('delay:' . $queue,$delayKey);
                  }
              }else{
                  self::redis()->zrem('delay:' . $queue,$delayKey);
              }
          }
      }
      return true;

    }
    
    
    public static function delay($queue, $class, $args = null,$delayTime = 60, $trackStatus = false)
    {
        require_once dirname(__FILE__) . '/Resque/Job.php';
        $result = Resque_Job::delay($queue, $class, $args, $delayTime,$trackStatus);
        if ($result) {
            Resque_Event::trigger('afterDelay', array(
                'class' => $class,
                'args'  => $args,
                'queue' => $queue,
            ));
        }

        return $result;
    }
    
}

调用

//Job中失败时

//$this->try_time  表示执行次数
if ($this->try_time < 3) {
    $this->job->delayJob(5 * 60); //5分钟后重新执行
}

//其他地方直接调用
Resque::delay($queue, $class, $args,$delayTime,$trackStatus);

solr介绍

发表于 2017-12-15

solr是什么呢？

一、Solr它是一种开放源码的、基于 Lucene Java 的搜索服务器，易于加入到 Web 应用程序中。

二、Solr 提供了层面搜索(就是统计)、命中醒目显示并且支持多种输出格式（包括XML/XSLT 和JSON等格式）。它易于安装和配置，而且附带了一个基于 HTTP 的
管理界面。Solr已经在众多大型的网站中使用，较为成熟和稳定。

三、Solr 包装并扩展了Lucene，所以Solr的基本上沿用了Lucene的相关术语。更重要的是，Solr 创建的索引与 Lucene 搜索引擎库完全兼容。

四、通过对Solr 进行适当的配置，某些情况下可能需要进行编码，Solr 可以阅读和使用构建到其他 Lucene 应用程序中的索引。

五、此外，很多 Lucene 工具（如Nutch、 Luke）也可以使用Solr 创建的索引。可以使用 Solr 的表现优异的基本搜索功能，也可以对它进行扩展从而满足企业的需要。

solr的优点

①高级的全文搜索功能；
②专为高通量的网络流量进行的优化；
③基于开放接口（XML和HTTP）的标准；
④综合的HTML管理界面；
⑤可伸缩性－能够有效地复制到另外一个Solr搜索服务器；
⑥使用XML配置达到灵活性和适配性；
⑦可扩展的插件体系。

配置文件

schema.xml /managed-schema

主键默认为id
1
<uniqueKey>id</uniqueKey>

字段类型

  <!--int 类型  -->
<fieldType name="int" class="solr.TrieIntField" docValues="true"precisionStep="0" positionIncrementGap="0"/>
<!--IKAnalyzer 分词器配置 -->
<fieldType name="text_ik" class="solr.TextField">
  <analyzer type="index" class="org.wltea.analyzer.lucene.IKAnalyzer"/>
  <analyzer type="query" class="org.wltea.analyzer.lucene.IKAnalyzer"/>
</fieldType>

添加字段

<!-- 主键必须存在 ，且当id 为主键时 类型必须为string-->
<field name="id" type="string" multiValued="false" indexed="true" required="true" stored="true"/>
  <field name="products_name" type="text_ik" indexed="true" stored="true"/>
    <!-- 动态字段  类似xxx_s的字段 无需添加名称， 类型默认为 string-->
  <dynamicField name="*_s" type="string" indexed="true" stored="true"/>

主键为比较项目且类型为 string

php 对接 solr

安装 solr扩展
- 下载地址 http://pecl.php.net/package/solr
  
  目前我们使用的是最新版 2.4.0
- phpinfo 中出现 solr 则表示安装成功
使用 solr
- 文档地址 http://php.net/manual/zh/book.solr.php
- 使用 SolrClient->qurey() 查询
- 使用 SolrClient->addDocument() 添加/更新文档

业务流程图

sequenceDiagram
user->>web:更新数据
web->>DB: 更新数据至数据库
web->>DB: 插入待更新solr数据
timer->>DB:获取要更新的数据
DB->>timer:返会更新数据
timer->>solr:推送更新数据至solr
user->>web:查询数据
web->>solr:获取数据
solr->>web:返回数据主键
web->>DB:通过主键获取数据
web->>user:展示数据

solr 功能

分词搜索

例如: 将搜索词 花王纸尿裤 拆分为 花王 和 纸尿裤 进行搜索

缺点:依赖分词器,分词结果可能与预想结果不同

模糊查询

通过相似度进行搜索，查询达到某个相似度的数据
权重

为不同字段设置权重，对不同字段设置不同权重,通过权重计算得出排序

查询范围

例如:查询价格区间
高亮显示

将匹配到关键词高亮显示
分组统计(Facet)

产品匹配到的数据有几个在类目1，几个在类目2 几个在某个属性组中
相似匹配

匹配与当前商品相似的商品
拼音检索

通过拼音查询中文数据

判断左右脑源码解读

发表于 2017-10-26

左右脑源码解读

最近看看你的左右脑年龄可以说是非常火，朋友圈有不少朋友人在玩，也看到有大牛对代码进行了解决，对此我也颇感兴趣来解读一番。
地址：[http://game.if122.com]

数据分析

查看其源码与网络请求，发现其发起了两次请求 js/index.json 与js/index_2.json。

首先是js/index.json ,返回结果为问题题目，共9个节点，第一个如下所示:

{
  "question":"这个男人的眼睛是在一条直线上吗？",
  "answer":{
    "是":"2",
    "否":"2"
  },
  "banner":"images/1_pWzVxxQ.jpg"
}

js/index_2.json 中返回年龄与应该展示共个4个节点每个节点中的word分别为a,b,c,d，第一个节点如下所示：

{"word":"a",
  "name":{
    "one":"25与27岁",
    "two":"39与22岁"
  },
  "output":
  {"img":"images/86DA2WyG5I5uXM1.jpg",
    "text":"李白属于灵活走位，多段位移，七进七出，秀你一脸，比较吃操作的英雄，因此玩家的意识要特别强，才能将该英雄玩的很好。需要多熟悉、多玩，对应新手来说，不是特别适合。"
  },
  "outputtwo":
  {"img":"images/ngeG9PYKAYsdqHtZ.jpg",
    "text":"李白属于灵活走位，多段位移，七进七出，秀你一脸，比较吃操作的英雄，因此玩家的意识要特别强，才能将该英雄玩的很好。需要多熟悉、多玩，对应新手来说，不是特别适合。"
  }
}

源码解析

查看js源码其中有两个随机数 ansrandom，ansrandom2，ansrandom2 先来看 ansrandom2 ，ansrandom2 的取值只与 .ad_after .ad_after_section 有关，查看源码，里面显示的是二维码等，所以与年龄判断无关,关键代码：
1
2
3
4
5
6
if(ansrandom2%4==0){
$('.ad_after .ad_after_section:eq(0)')
.css('display','flex')
.siblings()
.hide()
}

查看 ansrandom ,其结果为选择js/index_2.json 选中节点中的随机一个年龄

 if(ansrandom%2==1){
 	$('.result').prepend('<img src="'+result[index].output.img+'"class="result_img"/>')
 	.find('span').text(result[index].name.one);	
	$('title').text('原来我左右脑年龄是'+result[index].name.one+'，你也来看看');
 }else{
	
	//设置 outputtwo 略
}

接下来判断他是如何选中 js/index_2.json 中的节点的
关键代码:

// isNaN 来判断是否是最后一题
if(!isNaN($(this).data("math"))){
	//不是最后一题目 进入下一题
    var num_a=$(this).data("math")-1	$(".question:eq("+num_a+")").show().siblings().hide()
    json[num_a].banner?$(".banner").prop('src',json[num_a].banner):$(".banner").prop('src', json[0].banner)
}else{
	//最后一题目显示结果
	    
}

通过js/index.json 中的answer 选择来跳转下一题，而 answer 的中不同答案除了最后一个，都是相同而是 2，3,4,5,6,7,8,9 最后一个则分表为 abcd
所以前几题的结果都是无用的年龄判断只与之后一题有关，
最后一题的$(this).data("math");中放的分别是 abcd ，通过匹配abcd 在js/index_2.json 中选择出word 相同的节点，再通过 ansrandom 再在节点中随机选个年龄来展示。

结论

通过上一题的答案来选择下一题，但除最后一题外，跳转的题目都相同，故没有意义。
最后结果只与最后一题有关，在从你选择的答案中，从预设对应答案中选择选择一个来展示。
答案对应年龄：

a: 25与27岁或 39与22岁
b: 30与49岁或 9岁与26岁
c: 46岁与19岁或 33岁与5岁
d: 19与20岁

汽配兼容属性总表更新

发表于 2017-08-08

需求

从ERP获取汽配属性到站内

属性结构

每条兼容属性都挂在一个类目下
ERP 中的属性关联结构如下

graph LR
Model-->Make
Make-->Year
Make-->Engine
Make-->Trim

站内需要的结构如下

graph LR
Year-->Make
Make-->Model
Model-->Engine
Engine-->Trim

实现方案探究

使用ERP结构进行存储
```
建立5张表对应每个关联结构 记录条数与ERP中条数相同
```
缺点:
- 每次查询都要带上父级类目进行管理查询
- 结构复杂
- ERP结构变更
优点：
- 存储量小

使用站内结构

建立一张表 一条记录对应一个链式关系,如果ERP 中的条数为4n, 则站内对应条数n^4

缺点：

数据量大

优点：
- 结构简单
- 查询方便

实现方案

使用一条记录对应一个链式关系的方式存

结构如下

表字段名	字段
catetory	varchar(30)
year	smallint
make	varchar(30)
model	varchar(30)
egine	varchar(30)
trim	varchar(30)

实现

遍历获取ERP兼容属性，完成一个完成的关系链，插入数据库
优化

由于是 n^4 的记录条数，数据达到了恐怖的亿级
- 在结构上根据year水平分表，分成10个表，每个表只记录一个 year 位数的数据，插入数据是批量插入，采用如下形式，每千条插入一次
  1
  insert into table(year,make...) values (value1,value2...),(value1,value2...)
- 在查询上建立关联索引 year->make->model->engine->trim
- year层 union 查询生产缓存
- make 层是用索引后搜索的条数仍然达到了千万级别，仍使用缓存
- model,engine,trim 搜索的条数就在百万级以下查询速度感人，没有继续优化
成果
- innodb 下使用了将近50G的服务硬盘
- 更新一个类目耗时4,5个小时
  
  结束语
  到此完成了数据从ERP到站内的需求,总体来说，是用空间来换数据结构。
  虽然解决了对总表的更新，但感觉还有许多不足,期望日后能加以完善。

数独回溯算法

发表于 2017-08-07

数独回溯算法，计算未填写的九宫格

class bord {
    private $bord;

    public function __construct(array $bord) {
        $this->bord = $bord;
    }


    public function run() {
        $pos = $this->next();
      // var_dump($pos);
        if ($pos == false) {
            return true;
        };
        list($x, $y) = $pos;
//       echo $this->string($x,$y);
//       echo '<br/>';


        for ($v = 1; $v < 10; $v++) {
          //  var_dump($x, $y, $v);
           // var_dump($this->check($x, $y, $v));
          //echo $this->__toString();
            if ($this->check($x, $y, $v)) {
                $this->set($x, $y, $v);
                if($this->run()){
                    return true;
                };

            }


        }
        $this->release($x, $y);
     //   $this->run();

        return false;




    }

    public function next() {
        foreach ($this->bord as $x => $array) {
            foreach ($array as $y => $v) {
                if ($v == 0) {
                    return [$x, $y];
                }
            }
        }
        return false;
    }

    public function check($x, $y, $v) {
        foreach ($this->bord as  $x_a) {

            if ($x_a[$y] == $v) {
                return false;
            }
        }

        foreach ($this->bord[$x] as $y_v) {
           // var_dump($y_v);
            if ($y_v == $v) {
                return false;
            }
        }

        $x_b = intval(floor(($x-1)/3)*3 +1);

        $y_b = intval(floor(($y-1)/3)*3 +1);
       // var_dump('x',$x_b);var_dump('y',$y_b);

        for ($x = $x_b; $x < $x_b + 3; $x++) {
            for ($y = $y_b; $y < $y_b + 3; $y++) {
                if($this->bord[$x][$y] ==$v){
                    return false;
                }
            }
        }

        return true;
    }

    public function set($x, $y, $v) {
        $this->bord[$x][$y] = $v;
    }

    public function complete() {
        return $this->next() === false;
    }

    public function release($x, $y) {
        $this->bord[$x][$y] = 0;
    }
    public function string($x_r=null,$y_r=null) {
        $str = '';
        foreach ($this->bord as $x => $array) {
            foreach ($array as $y => $v) {
                if($x_r && $y_r && $x_r==$x && $y_r==$y){
                    $str .= '<span style="color:red">'.$v . '</span>&nbsp; &nbsp;  ';
                }else{
                    $str .= $v . '&nbsp; &nbsp;  ';
                }


            }
            $str .= '<br/>';
        }
        return $str;

    }
    public function __toString() {
        $str = '';
        foreach ($this->bord as $x => $array) {
            foreach ($array as $y => $v) {
                    $str .= $v . '&nbsp; &nbsp;';
            }
            $str .= '<br/>';
        }
        return $str;

    }
}

$bord  = new bord(array_fill(1,9,array_fill(1,9,0)));
$bord ->run();
echo $bord ;

js 中参数的获取

发表于 2017-08-07

js 中参数的获取

柯里化(currying)

function foo(x) {
    return function (y) {
        console.info(x+y);
    }
}
foo(5)(6);// 11 ，如果方法中没有改变量，则向上层寻找
new foo; //注意，此时返回的不是foo 对象（注释1）

arguments，获取方法传入的所有值（场景：参数不固定）
1
2
3
4
function foo() {
console.info(arguments)
}
foo(1,2);//[1,2]

…x 类似php 5.6 特性中的参数获取

function foo (...x){
    for(var z of x){
         console.warn(z)
       }
}

注释

1 . 当方法返回一个对象，或一个方法时,实例化这个方法时，得到他的返回值的实例

前言

功能使用

说明

使用步骤

监控

solr是什么呢？

solr的优点

配置文件

php 对接 solr

业务流程图

solr 功能

左右脑源码解读

数据分析

源码解析

结论

需求

属性结构

实现方案探究

实现方案

结束语

数独 回溯算法，计算未填写的 九宫格

js 中参数的获取

注释

数独回溯算法，计算未填写的九宫格