Elasticsearch Query DsL查询


Query DsL查询

一 Elasticsearch简介

Elasticsearch 是一个开源的搜索引擎,Elasticsearch 使用 Java 编写的,它的内部使用 Lucene 做索引与搜索,但是它的目的是使全文检索变得简单, 通过隐藏 Lucene 的复杂性,取而代之的提供一套简单一致的 RESTful API。

  • 一个分布式的实时文档存储,每个字段 可以被索引与搜索
  • 一个分布式实时分析搜索引擎
  • 能胜任上百个服务节点的扩展,并支持 PB 级别的结构化或者非结构化数据

 

二 安装并运行

已经在其他文档中详细介绍,此次仅做简单步骤介绍

# 安装:
$   wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.13.3-linux-x86_64.tar.gz
$   tar -xzf elasticsearch-7.13.3-linux-x86_64.tar.gz 
$   cd elasticsearch-7.13.3/
# 运行
sh bin/elasticsearch

# 访问
$ curl http://192.168.3.14:9200/ 
{
  "name" : "87DNZWU",
  "cluster_name" : "elasticsearch",
  "cluster_uuid" : "e3A3l85MSZuZlRhxj6IB2w",
  "version" : {
    "number" : "6.7.0",
    "build_flavor" : "default",
    "build_type" : "zip",
    "build_hash" : "8453f77",
    "build_date" : "2019-03-21T15:32:29.844721Z",
    "build_snapshot" : false,
    "lucene_version" : "7.7.0",
    "minimum_wire_compatibility_version" : "5.6.0",
    "minimum_index_compatibility_version" : "5.0.0"
  },
  "tagline" : "You Know, for Search"
}

 

三 Query DSL 基本结构

查询表达式(Query DSL)是一种非常灵活又富有表现力的查询语言, Elasticsearch 使用它可以以简单的 JSON 接口来展现 Lucene 功能的绝大部分

// 查询
GET /_search  // 查找整个ES中所有索引的内容
{
  "query": {}, //具体的查询语句对象
  "from": 0,   //从第几条数据开始返回
  "size": 100, //返回的条数 默认ES最多返回10000条
  "highlight": { //高亮
    "pre_tags": {}, //高亮内容的前面标签 一般都是html比如<b> <p>这种
    "post_tags": {},//高亮内容的后面标签 一般都是html比如</b> </p>这种
    "fields": { //需要高亮的字段
    }
  },
  "sort": [{ //排序
    "FIELD": { //排序的字段(需要填上具体的字段名)
      "order": "desc"
    }
  }],
  "_source": "{field}" //指定返回的字段
}

// 结果
{
    "took": 350,  // 整个搜索请求消耗了多少毫秒
    "timed_out": false, // 表示本次查询是否超时,如果为true也会返回结果,只是数据可能不完整
    "_shards": { //  显示查询中参与的分片信息,成功多少分片失败多少分片等
        "total": 5, 
        "successful": 5,
        "skipped": 0,
        "failed": 0
    },
    "hits": {
        "total": 5, // 匹配到的文档总数
        "max_score": 1, // 为文档中所有_score的最大值
        "hits": [
            {
                "_index": "mysql-shop_trades-order_item_label_binds",
                "_type": "doc",
                "_id": "591935",
                "_score": 1,
                "_source": {
                    "id": 591935,
                    "updated_at": "2021-05-20T06:26:09.000Z",
                    "@version": "1",
                    "bind_item_label_id": 729,
                    "label_type": "brand",
                    "created_at": "2021-05-20T06:26:09.000Z",
                    "@timestamp": "2021-07-07T07:31:36.262Z",
                    "is_deleted": 0,
                    "table_name": "order_item_label_binds",
                    "bar_code": "6907925004486"
                }
            }
        ]
    }
}

 

四 指定索引搜索

 

上述查询会搜索ES中的所有索引,但通常只需要去固定一个或几个索引中搜索,搜索全部无疑会造成资源的浪费,在ES中可以通过以下几种方法来指定索引

  1. 指定一个固定的索引,ops-coffee-nginx-2019.05.15为索引名字
GET /mysql-shop_trades-order_statics/_search

以上表示在mysql-shop_trades-order_statics索引下查找数据

  1. 指定多个固定索引,多个索引名字用逗号分割
GET /mysql-shop_trades-order_statics,mysql-shop_trades-order_item_labels/_search

  1. 用*号匹配,在匹配到的所有索引下查找数据
GET /mysql-shop_trades-*/_search

这里也可以用逗号分割多个匹配索引

五 DSL查询

1、筛选字段

// 筛选_source的数据,单个字段
GET /_search
{
  "_source": "bar_code",  
  "query": {}
}

// 筛选_source的数据,多个字段
{
  "_source": {
    "includes": ["store_id", "sku_id"]
  },
  "query": {}
}

// 对字段进行转换
{
  "docvalue_fields": [
    {
      "field": "updated_at",
      "format": "yyyy-MM-dd HH:mm:ss"
    },
    {
      "field": "num",
      "format": "long" // 没有作用,懵逼...
    }
  ], 
  "query": {}
}

 

2、多条件查询 (where)

  1. constant_score:装另一个查询的查询,固定分数查询,支持filter查询,不支持match查询:
    {
        "constant_score": {
            "filter": {
                "match": {
                    "name": "小米"
                }
            },
            "boost": 10
        }
    }
    

     

  2. bool:主要与其他关键字组合使用,多条件的查询必须要用bool包在外层,然后再根据具体的业务来拼接。

{
  "query": {
    "bool": {
      "should": [{}], //满足其中一个对象查询条件就行 像sql里的or
      "must": [{}],   //必须满足所有对象的查询条件 就像sql里的and
      "must_not": [{}] //必须不满足所有对象的查询条件 就像sql里的and !=
    }
  }
}

 

  1. must: 类似于SQL中的AND,必须包含
  2. must_not: 类似于SQL中的NOT,必须不包含
  3. should: 满足这些条件中的任何条件都会增加评分_score,不满足也不影响,should只会影响查询结果的_score值,并不会影响结果的内容
  4. filter: 与must相似,但不会对结果进行相关性评分_score,大多数情况下我们对于日志的需求都无相关性的要求,所以建议查询的过程中多用filter

3、group by:

ES本身没有group关键词搜索,但支持聚合查询,,需要使用关键字aggs

// 单个字段 group by
{
  "query":{},//这里省略你的查询条件
  "aggs": {
    "age_group": {//这个是指你要返回字段名
      "terms": { //这里还可以用其它关键词 这里terms才能实现group by效果
        "field": "age",//groupby的字段
        "size":1 //返回的条数 相当于group by limit
      }
    }
  }
}

// 多字段group by (如 group by sku_id,store_id)
// 方法一:script
{
  "query":{},
  "aggs": {
    "age_group": {
      "terms": {
        "script":{
          "source": """ 's' + doc['store_id'] + '_s' + doc['sku_id'] """,
          "lang": "painless"
        },
        "size": 10
      }
    }
  }
}

// 方法二:copy to
1. 设置mapping中的多个字段,copy_to 为同一个字段(skuId_storeId)
2. 搜索新字段
{
  "query":{},
  "aggs": {
    "list": {
      "terms": { 
        "field": "skuId_storeId
        "size":1
      }
    }
  }
}

// 方法三:multi_terms (使用高版本,目前6.7不支持)
{
  "aggs": {
    "genres_and_products": {
      "multi_terms": {
        "terms": [{
          "field": "genre" 
        }, {
          "field": "product"
        }]
      }
    }
  }
}

4、order by

order by:注意日期格式和数值格式才支持排序;文本不支持,如果要排序, 需把字段设置为not analysis

// 单排序
{
    "query": {
        "sort": {
            "id": "desc"
        }
    }
}

// avg按照平均值排序
{
    "query": {
        "sort": [
            {
                "id": "desc"
            },
            {
                "price": {
                    "order": "asc",
                    "mode": "avg"
                }
            }
        ]
    }
}

5、count(distinct)

{
  "query":{},
  "aggs": {
    "total_sku_id": {
      "cardinality":{ "field": "sku_id"}
    },
     "total_entity_store_id": { // 非数字类型,无法使用field排序,可以对field增加fieldData = true,或者对field.keyword排序,建议使用后者,高效内存消耗低
      "cardinality":{ "field": "entity_store_id.keyword"}
    }
  }
}

 

6、SUM

{
  "query":{},
  "aggs": {
    "total_pay_num": {
      "sum": {"field": "num"}
    },
     "total_cost_fee": {
      "sum": {"field": "cost_fee"}
    }
  }
}

 

7、distinct :

select distinct(id) from table

{
  "query":{},
  "collapse": {
      "field": "id" //你需要distinct的字段
   }, 
}

 

8、limit

1. 分页:
    1. form:从第几个开始查询,最开始是0
    2. size:即limit
    3. 使用size,size最大可获取数量是xx个
2. 获取所有数据的三种方式
    1. scroll深度滚动需要根据scroll_id和循环取,取完后,需删除scroll,减小内存开销(深度滚动高效,用于处理大量数据,不适合实时获取)
    2.调整索引index.max_result_window的大小,默认10000 (大小与堆内存成正比,这个限制内存)
    3.search_after:请求需增加前一个结果的排序,(实时游标,可根据索引更新和删除而改变 )
    4。 如果是group by查询获取所有数据, 获取需要使用到cardinality查询预估总数,再使用partition、num_partitions分区依次获取数据

9、搜索关键字

  1. match:自定字段,根据字段关键字进行搜索,会分割关键词,匹配到含有一个多多个词的匹配
  2. query_string:全文搜索
  3. match_phrase:不分割关键词 {"match_phrase": {"name":"婴幼儿奶粉"}}
  4. term: 类似SQL where field = x,主要用于数字匹配;如果要匹配文本,会自动分词,不能精准查询,需把字段设置成not analysis
    {
      "query": {
        "term": {"bind_item_label_id": 729}
      }
    }
    

     

  5. terms: 类似SQL where field in (x,x),主要用于数字匹配,
    {
      "query": {
        "terms": {"bind_item_label_id": [703,729]}
      }
    }
    

     

  6. range:: 查询价格在1000-2000的商品
{
    "query": {
        "range": {
            "price": {
                "gte": 1000,
                "lte": 2000
            }
        }
    }
}

 

  1. filter:判断文档是否满足条件
{
    "query": {
        "bool": {
            "filter": {
                "term": {
                    "price": 1999
                }
            }
        }
    }
}

Elasticsearch:Aggregation

  1. metric:度量聚合,主要针对number类型的数据,需要es做较多的计算工作(类似SQL的SUM、MAX、AVG、MIN、Cardinality、stats<属于多值分析>等)
  2. bucket:桶聚合,划分不同步的桶,将数据分配到不同的桶,(类似SQL中的group by)
  3. Pipeline Aggregation:管道分析类型,对其他聚合结果进行二次聚合
  4. Matrix Aggregation:矩阵分析类型,支持对多个字段的操作并提供一个结果矩阵

term aggregation

  • size 可以通过size返回top size的文档,该术语聚合针对顶层术语(不包含嵌套词根),其搜索过程是将请求向所有分点发送请求,每个分片节点返回size条数据,然后聚合所有分片的结果(会对各分片返回的同样词根的数数值进行相加),最终从中挑选size条记录返回给客户端。从这个过程也可以看出,其结果并不是准确的,而是一个近似值。
  • Shard Size 为了提高该聚合的精确度,可以通过shard_size参数设置协调节点向各个分片请求的词根个数,然后在协调节点进行聚合,最后只返回size个词根给到客户端,shard_size >= size,如果shard_size设置小于size,ES会自动将其设置为size,默认情况下shard_size建议设置为(1.5 * size + 10)。
// 单个字段 group by
{
  "query":{},//这里省略你的查询条件
  "aggs": {
    "age_group": {//这个是指你要返回字段名
      "terms": { //这里还可以用其它关键词 这里terms才能实现group by效果
        "field": "age",//groupby的字段
        "size":1 //返回的条数 相当于group by limit
      }
    }
  }
}

// 返回结果格式
{
  ...
  "aggregations" : {
      "list" : {
        "doc_count_error_upper_bound" : 0, // 该值表示未进入最终术语列表的术语的最大潜在文档计数
        "sum_other_doc_count" : 90 // 该值表示未进入最终术语列表的术语的最大潜在文档计数
        "buckets" : [ // 返回doc_count排名最前的10个,受size参数的影响
          {
            "key" : "1",
            "doc_count" : 24,
            "total_refund_fee" : {
              "value" : 0.0
            },
            "total_cost_fee" : {
              "value" : -14976.0
            },
          }
        ]
      }
    }
    }
}

 

 

match_phrase :查询分析文本,创建词组查询

https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl-match-query-phrase.html#query-dsl-match-query-phrase

 

 

举例子

GET mysql-shop_trades-order_item_label_binds/_search/?scroll=1m
{
  "docvalue_fields": [
    {
      "field": "updated_at",
      "format": "yyyy-MM-dd HH:mm:ss"
    }
  ], 
  "size": 1000,
  "sort": {"id":"desc"},
  "query": {
    "bool": {
      "must": [
        {"match": {"is_deleted": 0}},
        {"match": {"label_type": "brand"}},
        {
          "constant_score": {
            "filter": {
              "terms": {
                "bind_item_label_id": [703, 2, 729]
                }
              }
            }
        }
      ]
    }
    }
}




GET mysql-shop_trades-order_item_label_binds/_search/?scroll=1m
{
  "_source": "bar_code", 
  "query": {
    "bool": {
      "filter": [
        {"match": {"is_deleted": 0}},
        {"match_phrase": {"label_type": "brand"}},
        {"terms": {"bind_item_label_id": [703, 2, 729]}}
      ]
    }
  },
  "aggs": {
    "bar_code_group": {
      "terms": {
        "field": "bar_code.keyword",
        "size": 10 
      }
    }
  }
}


GET mysql-shop_trades-order_item_label_binds,mysql-shop_trades-order_statics/_search
{
  "query": {
    "bool": {
      "filter": [
        {"match_phrase": {"sys_name": "yiqigou"}},
        {"range": {"num": {"lte": 2000}}},
        {"range": {"return_num": {"gte": -1000}}},
        {"range": {"total_price": {"lte": 1000000}}},
          {"match": {"id": 60}},
        {"term": {"order_type": 0}},
        {"term": {"item_type": 0}},
        {"range": {"date": {
          "gte": "2020-01-21 00:00:00",
          "lte": "2021-07-22 00:00:00",
          "format": "yyyy-MM-dd HH:mm:ss",
          "time_zone": "+08:00"
        }}}
      ],
      "must_not": [
        {"terms": {"store_id": [165]}}
      ]
    }
  }
}

 

 

设置fieldData

// 第一步,创建索引 (如果已经有索引,直接看第二步)
PUT mysql-shop_trades-order_statics2
{
  "mappings": {
    "_doc": {
      "properties": {
        "entity_store_id": { 
          "type": "text",
          "fields": {
            "keyword": { 
              "type": "keyword"
            }
          }
        }
      }
    }
  }
}

// 第二步 设置fieldData为true
PUT mysql-shop_trades-order_statics/_mapping/_doc
{
  "properties": {
    "entity_store_id": {
      "type":     "text",
      "fielddata": true
    }
  }
}

// 第三步 可以查看该索引的Mapping结构,fieldData是否加上去
{
  "mapping": {
    "doc": {
      "properties": {
        "entity_store_id": {
          "type": "text",
          "fields": {
            "keyword": {
              "type": "keyword",
              "ignore_above": 256
            }
          },
          "fielddata": true
        }
      }
    }
  }
}

延伸

 

 

设置 max_result_window

PUT /mysql-shop_trades-order_statics/_settings
{
  "index": {
    "max_result_window": 100000
  }
}

 

Spring Boot 入门

一、开发基础

  • Java基础(两到三小时过一遍)
  • Java开发环境配置必须使用JDK1.8
  • IDE安装(优先使用IntelliJ IDEA)

二、名词解释

  • Spring:JAVA开发应用框架
  • Spring Boot:用来简化Spring应用的初始搭建以及开发过程的配置框架
  • Maven:Java项目构建工具,成熟的项目
  • Gradle:更简洁的Java项目构建工具,吸收了旧构建工具的优点。
  • JPA:是Sun官方提出的Java持久化规范,即数据库操作规范。
  • Hibernate:Hibernate是一个ORM框架,是JPA的默认实现方式,一般说JPA都是指Hibernate。
  • Mybatis:Mybatis是一个轻便的ORM框架。
  • Spring data jpa:是Spring基于ORM框架、JPA规范的基础上封装的一套JPA应用框架

三、新项目流程

  1. 新建gradle项目
    • File->new->Prroject->Spring Initializr

    • 填写Group、Artifact选择Gradle Project项目生成

    • 可以直接在生成项目的时候选择对应需要安装的插件,如:web、jpa、mybatis等,也可以在项目初始化完成之后在build.gradle中添加/配置

  2. 配置build.gradle(位于根目录)

    plugins {
        id 'org.springframework.boot' version '2.1.3.RELEASE'
        id 'java'
    }
    
    apply plugin: 'io.spring.dependency-management'
    
    group = 'com.duomai'
    version = '0.0.1-SNAPSHOT'
    sourceCompatibility = '1.8' // JDK最大兼容版本
    
    repositories {
        mavenCentral()
    }
    
    dependencies {
        // Spring Boot JPA 组件
        implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
        // Spring Boot Web组件
        implementation 'org.springframework.boot:spring-boot-starter-web'
        // Mybatis插件,注意暂时使用**1.1.1**版本,高版本的运行好像有问题
        implementation 'org.mybatis.spring.boot:mybatis-spring-boot-starter:1.1.1'
        // Mybatis分页插件
        implementation group: 'com.github.pagehelper', name: 'pagehelper-spring-boot-starter', version: '1.2.10'
        runtimeOnly 'org.springframework.boot:spring-boot-devtools'
        runtimeOnly 'mysql:mysql-connector-java'
        testImplementation 'org.springframework.boot:spring-boot-starter-test'
    }
    
    

    修改了build.gradle后,idea会自动安装/更新依赖包。
    参考:gradle官网Spring Boot Web服务搭建Spring Boot Mysql使用Spring Boot JPA使用

  3. 项目基础配置(位于src/resources/application.properties

    #运行配置
    server.port=9000
    #数据格式配置
    spring.jackson.time-zone=GMT+8 // 设置接口返回时区为东八区
    spring.jackson.date-format=yyyy-MM-dd HH:mm:ss // 自动将接口返回中的日期格式转换为标准格式
    #数据库连接配置
    spring.datasource.url=jdbc:mysql://192.168.0.235:3355/shop_balances?serverTimezone=Asia/Shanghai&tinyInt1isBit=false // serverTimezone选择Mysql东八区,tinyInt1isBit禁止Mysql自动将tinyint(1)类型数据映射为boolean类型
    spring.datasource.username=cishop
    spring.datasource.password=fuyuan1906
    #log配置
    logging.path=E:/java/demo/balance_card/log
    logging.level.com.favorites=DEBUG
    #logging.level.org.springframework.web=INFO
    logging.level.org.hibernate=ERROR
    #mybatis设置
    mybatis.type-aliases-package=com.duomai.balance_card.Model.Mapper
    mybatis.configuration.map-underscore-to-camel-case=true
    logging.level.com.duomai.balance_card.Model.Mapper=DEBUG
    #pagehelper插件设置
    pagehelper.helperDialect=mysql
    pagehelper.reasonable=false
    pagehelper.supportMethodsArguments=true
    pagehelper.params=pageNum=page;pageSize=limit
    #jpa 设置
    spring.jpa.properties.hibernate.hbm2ddl.auto=update
    spring.jpa.properties.hibernate.dialect=org.hibernate.dialect.MySQL5InnoDBDialect
    spring.jpa.show-sql= true
    
  4. 项目启动
    • IDEA启动,运行[Prefix]Application.java文件即可

    • 命令行启动

      • Maven项目
        在springboot的应用的根目录下运行 mvn spring-boot:run

      • Grdale项目
        在springboot的应用的根目录下运行 gradle bootRungradlew bootRun
        (前者是使用本地的gradle版本运行,后者是使用代码仓库中的gradle运行)

    • 打包构建可执行文件运行

  5. 开发环境热更新设置
    热更新不是很好用,有一定的延迟时间

四、项目文件分层解析

// 业务代码 src/main
java 
    com
        duomai
            balance_card
                [Prefix]Application.java // 项目启动文件,可以做一些全局设置,如时区设置、Mapper扫描等
                Config // 配置类,用于注册一些全局配置,如拦截器注册等
                Middleware // 中间件,实现AOP功能
                Controller // 控制器,主要做路由功能
                    xxxController.java
                    BaseErrorController.java // 路由匹配失败时使用的控制器
                Service // 业务代码
                Model // 目录主要用于实体与数据访问层
                    Entity // 数据表实体类
                    Repository // JPA数据仓库
                    Mapper // Mybatis映射文件
                    Provider // Mapper的Sql生成器
                Library // 库类,存放公共类文件/纯定义文件等
                    ApiReturnDefines.java // 接口返回定义
                    ExceptionErrprDefines.java // 异常监听层定义
                Helper // 辅助函数类文件
                OutPut // 接口输出层
                    ApiResult.java // 最终的接口输出格式
                    AiReturn.java // 快速生成ApiResult类,供外部调用
                Exception // 统一的异常处理
                    ControllerHandler // 路由层异常监听
                    SqlHandler // 数据库层异常监听
// 配置项
resources
    appliaction.properties // 项目配置文件

五、控制器中间件

  1. 一般使用Spring过滤器或拦截器实现AOP切面编程

  2. 过滤器和拦截器的对比

    • 作用域不同
      过滤器依赖于servlet容器,只能在 servlet容器,web环境下使用。
      拦截器依赖于spring容器,可以在spring容器中调用,不管此时Spring处于什么环境。
    • 细粒度的不同
      过滤器的控制比较粗,只能在请求进来时进行处理,对请求和响应进行包装。
      拦截器提供更精细的控制,可以分为controller对请求处理之前、渲染视图之后、请求处理之后三个切面。
    • 中断链执行的难易程度不同
      拦截器可以 preHandle方法内返回 false 进行中断。
      过滤器就比较复杂,需要处理请求和响应对象来引发中断,需要额外的动作,比如将用户重定向到错误页面。
  3. 拦截器的使用
    • 编写自定义拦截器(Middleware/ControllerInterceptor.java
    public class ControllerInterceptor implements HandlerInterceptor {
        private Logger logger = LoggerFactory.getLogger(ControllerInterceptor.class);
        @Override
        public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
            logger.info("preHandle....");
            // token校验等
            String token = request.getHeader("token");
            //     Common.sendJson(response, ApiReturn.fail(1001, "token验证失败"));
            //      return false;
            return true;
        }
    
        @Override
        public void postHandle(HttpServletRequest request, HttpServletResponse response, Object handler, ModelAndView modelAndView) throws Exception {
            logger.info("postHandle...");
        }
    
        @Override
        public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {
            // 接口日志记录等
            logger.info("afterCompletion...");
        }
     ```
    
    **说明**:
    preHandle:对客户端发过来的请求进行前置处理,如果方法返回true,继续执行后续操作,如果返回false,执行中断请求处理,请求不会发送到Controller。可以在这里校验一些权限信息,如token等,校验失败直接以JSON格式返回请求。
    
    postHandler:在请求进行处理后执行,也就是在Controller方法调用之后处理,前提是preHandle方法返回true。具体来说,postHandler方法会在DispatcherServlet进行视图返回渲染前被调用。
    
    afterCompletion: 该方法在整个请求结束之后执行,前提依然是preHandle方法的返回值为true。
    
    • 注册拦截器(Config/InterceptorConfig.java
    @Configuration
    public class InterceptorConfig implements WebMvcConfigurer {
        @Override
        // 核心方法
        public void addInterceptors(InterceptorRegistry registry) {
            registry.addInterceptor(ControllerInterceptor())
                    //配置拦截规则
                    .addPathPatterns("/**")
                    .order(1);
    
            // 多个拦截器按上述方法持续注册即可,同时也可以设置order值,从小到大执行。
        }
    
        @Bean
        public HandlerInterceptor ControllerInterceptor() {
            return new ControllerInterceptor();
        }
    }
    

六、Controller层

  1. 定义控制器文件
    @RestController
    在类文件头部定义,标明为控制器文件,且输出格式为JSON
    
  2. 路由和参数
    1. 定义路由名称,接收方法
      例:@RequestMapping(value = "/get", method = {RequestMethod.GET, RequestMethod.POST})
      
        可选参数:
      
        value:路由名称
      
        method:指定请求的method类型, GET、POST、PUT、DELETE、PATCH等,可多选
      
        consumes:指定处理请求的提交内容类型(Content-Type),例如application/json, text/html
      
        produces: 指定返回的内容类型,仅当request请求头中的(Accept)类型中包含该指定类型才返回
      
        params: 指定request中必须包含某些参数值
      
        headers: 指定request中必须包含某些指定的header值
      
    2. 参数接收
      例:@RequestParam(value = "fields", required = false, defaultValue = "*") String fields
      
          value:参数名称
      
          defaultValue:默认值
      
          required:是否是必要参数
      
    3. 自定义错误路由
      1. 在Controller层中添加BaseErrorController.java文件,用于监听路由匹配失败的情况
        @Controller
        public class BaseErrorController implements ErrorController {
            @Override
            public String getErrorPath() {
                System.out.print("错误页面");
                return "error/error";
        }
        
            @RequestMapping(value = "/error")
            public void error() throws Exception {
                throw new Exception("路由匹配失败");
            }
        }
        
      2. 在Exception文件夹中添加ControllerHandler.java,用于捕获路由报错并输出。
        @RestControllerAdvice
        public class ControllerHandler {
            // 缺少必选参数
            @ExceptionHandler({MissingServletRequestParameterException.class})
            @ResponseBody
            public ApiResult requestMissingServletRequest(MissingServletRequestParameterException e){
                return ApiReturn.fail(ExceptionErrorDefines.RequestMissingServletRequest, e.getMessage());
            }
        }
        未解决:抛出异常后访问404页面运行环境会报错,但是页面正常  
        

        参考:https://www.jianshu.com/p/393f70b55b1b

  3. Service层调用

    1. Service类成员注入
      • 使用@Autowired修饰符进行依赖注入
        @Autowired
        private final CardService cardService;
        
      • 用构造函数来做注入类成员(推荐使用)
        private StoreBalanceCardsRepository cardsRepository;
        public CardController(StoreBalanceCardsRepository cardsRepository) {
            this.cardsRepository = cardsRepository;
        }
        **注**:
        IntelliJ IDEA使用依赖注入会有IDE报错,但不影响实际编译运行,如需去除报错提示,需要在Dao层(Respository/Mapper)类开头添加注解 `@Repository`
        
    2. 调用
      cardService.get(id, fields);
      

七、Service层

  1. 定义Service文件
    @Service
    在类文件头部定义,标明为Service文件
    
  2. 注入Model层操作文件
    private final StoreBalanceCardsRepository storeBalanceCardsRepository;
    private final StoreBalanceCardsMapper storeBalanceCardsMapper;
    
        CardService(StoreBalanceCardsRepository storeBalanceCardsRepository, StoreBalanceCardsMapper storeBalanceCardsMapper) {
            this.storeBalanceCardsRepository = storeBalanceCardsRepository;
            this.storeBalanceCardsMapper = storeBalanceCardsMapper;
        }
    
  3. 调用Model层文件

八、Model层

  1. Repository

    • Spring中概念,概念类似于数据仓库,是Spring data jpa的实现。居于业务层和数据层之间,将两者隔离开来,在它的内部封装了数据查询和存储的逻辑。

    • Repository和传统意义上的DAO的区别:
      Repository蕴含着真正的OO概念,即一个数据仓库角色,负责所有对象的持久化管理。DAO则没有摆脱数据的影子,仍然停留在数据操作的层面上。Repository是相对对象而言,DAO则是相对数据库而言,虽然可能是同一个东西,但侧重点完全不同。

  2. Mapper

    存放Mybatis数据库关系映射方法

  3. Provider
    为Mapper层提供的SQL生成器,即将SQL的生成与映射解耦。

  4. Entity

    • 根据表结构自动生成实体类
      **注意**:
      i、多次自动生成不会覆盖,如需更新需要把旧的实体类文件删除
      
      ii、有需要的话可以在实体类文件右击generate一键生成构造函数、set get方法、@Autowired等
      
      iii、自动生成的实体类中datetime类型的字段会被转为Java的timestamp类型数据,存储的时候也使用timestamp类型即可
      
      iv、生成出来的实体类catalog = "" 报错,是IDE的报错,不影响使用
      

      参考文档:https://blog.csdn.net/chenju05244554/article/details/1009142081

  5. Hibernate 和 Mybatis 的对比

    • Hibernate优势:
      i、DAO层开发比MyBatis简单,Mybatis需要维护SQL和结果映射。
      
      ii、对对象的维护和缓存要比MyBatis好,对增删改查的对象的维护要方便。
      
      iii、数据库移植性很好,MyBatis的数据库移植性不好,不同的数据库需要写不同SQL。
      
      iv、有更好的二级缓存机制,可以使用第三方缓存。MyBatis本身提供的缓存机制不佳。
      
    • Mybatis优势:
      i、MyBatis可以进行更为细致的SQL优化,可以减少查询字段。
      
      ii、MyBatis容易掌握,而Hibernate门槛较高。
      
    • 选用Mybatis的原因
      i、Hibernate无法满足动态获取部分字段的需求,即使是使用Hibernate提供的原始SQL也无法实现
      
      ii、Hibernate的JPA查询只适用于一些简单的情况,如遇到复杂的SQL,Repository中的方法名会很长。这时候又将回到Hibernate的自定义SQL查询,即原生SQL。
      
      iii、Hibernate的维护成本比MyBatis高很多,MyBatis的SQL生成完全取决于开发者,所以SQL修改、维护、优化会比较便利。
      

九、MyBatis使用

  1. 引入MyBatis以及pagehelper分页插件
    dependencies {
        implementation 'org.mybatis.spring.boot:mybatis-spring-boot-starter:1.1.1'
        implementation group: 'com.github.pagehelper', name: 'pagehelper-spring-boot-starter', version: '1.2.10'
    }
    
  2. application.properties 添加相关配置
    #mybatis设置
    mybatis.type-aliases-package=com.duomai.balance_card.Model.Mapper // 项目中Mapper存放的包
    mybatis.configuration.map-underscore-to-camel-case=true // 自动将sql字段下划线转为驼峰,可以保证取出的数据格式就是数据库中存储的格式
    logging.level.com.duomai.balance_card.Model.Mapper=DEBUG // 开启DEBUG模式,用于开发环境,记录执行的SQL。
    #pagehelper插件设置
    pagehelper.helperDialect=mysql // 指定分页插件使用哪种数据库
    pagehelper.reasonable=false // 分页合理化参数,默认值为false。当该参数设置为 true 时,pageNum<=0 时会查询第一页, pageNum>pages(超过总数时),会查询最后一页。默认false 时,直接根据参数进行查询。
    pagehelper.supportMethodsArguments=true // 支持通过 Mapper 接口参数来传递分页参数,默认值false,分页插件会从查询方法的参数值中,自动根据上面 params 配置的字段中取值,查找到合适的值时就会自动分页,设置后无需手动启用分页插件。
    pagehelper.params=pageNum=page;pageSize=limit //  自定义Mapper 接口参数来传递分页参数的参数名
    
  3. 添加Mapper类扫描的两种方式
    • 在启动类中添加对mapper包扫描@MapperScan(推荐使用)
      @SpringBootApplication
      @MapperScan("com.duomai.balance_card.Model.Mapper")
      
    • 在具体的Mapper类上面添加注解 @Mapper

  4. Mapper类SQL用例

    public interface StoreBalanceCardsMapper {
        // select用例
        @SelectProvider(type = StoreBalanceCardsSqlBuilder.class, method = "getById")
        List<Map> getById(@Param("id") int id, @Param("fields") String fields, @Param("page") int page, @Param("limit") int limit);
    
        // 列表
        @SelectProvider(type = StoreBalanceCardsSqlBuilder.class, method = "list")
        Page<Map> list(@Param("fields") String fields, @Param("page") int page, @Param("limit") int limit);
    
        // insert用例
        @InsertProvider(type = StoreBalanceCardsSqlBuilder.class, method = "add")
        @Options(useGeneratedKeys=true, keyProperty="storeBalanceCard.id", keyColumn="id")
        int add(@Param("storeBalanceCard") StoreBalanceCards storeBalanceCard);
    
        // update用例
        @UpdateProvider(type = StoreBalanceCardsSqlBuilder.class, method = "updateName")
        @Options(useGeneratedKeys=true, keyProperty="storeBalanceCard.id", keyColumn="id")
        int updateName(@Param("storeBalanceCard") StoreBalanceCards storeBalanceCard, @Param("limit") int limit);
    
        // delete用例
        @DeleteProvider(type = StoreBalanceCardsSqlBuilder.class, method = "deleteById")
        int delete(@Param("id") int id, @Param("limit") int limit);
    }
    
    • 通过Provider的方式动态获取SQL

    • 需要分页时在具体的方法后多加page、limit参数,自动实现分页

      // Service层调用
      List storeBalanceCard = storeBalanceCardsMapper.list(fields, page, limit);
      // Mapper层实现
      Page<Map> list(@Param("fields") String fields, @Param("page") int page, @Param("limit") int limit);
      
    • 分页组件仅可用于查询,不可用于更新/删除,更新/删除需要另外实现

  5. Providerle类SQL用例

    public class StoreBalanceCardsSqlBuilder {
        private static final String STORE_BALANCE_CARDS = "store_balance_cards";
        public static String getById(int id, String fields) {
            return new SQL(){ { 
                SELECT(fields);
                FROM(STORE_BALANCE_CARDS);
                WHERE("store_balance_cards.id = #{id}");
            } }.toString();
        }
    
        public static String list(String fields) {
            return new SQL(){ {
                SELECT(fields);
                FROM(STORE_BALANCE_CARDS);
            } }.toString();
        }
    
        public static String add(@Param("storeBalanceCard") StoreBalanceCards storeBalanceCard) {
            return new SQL(){ {
                INSERT_INTO(STORE_BALANCE_CARDS);
                VALUES("sys_name", "#{storeBalanceCard.sysName}");
                VALUES("store_id", "#{storeBalanceCard.storeId}");
                VALUES("entity_store_id", "#{storeBalanceCard.entityStoreId}");
                VALUES("name", "#{storeBalanceCard.name}");
                VALUES("type", "#{storeBalanceCard.type}");
                VALUES("item_id", "#{storeBalanceCard.itemId}");
                VALUES("photo", "#{storeBalanceCard.photo}");
                VALUES("state", "#{storeBalanceCard.state}");
                VALUES("upgrade", "#{storeBalanceCard.upgrade}");
                VALUES("recharge_discount", "#{storeBalanceCard.rechargeDiscount}");
                VALUES("updated_at", "#{storeBalanceCard.updatedAt}");
                VALUES("created_at", "#{storeBalanceCard.createdAt}");
            } }.toString();
        }
    
        public static String updateName(@Param("storeBalanceCard") StoreBalanceCards storeBalanceCard, @Param("limit") int limit) {
            return new SQL(){ {
                UPDATE(STORE_BALANCE_CARDS);
                SET("name=#{storeBalanceCard.name}");
                if (storeBalanceCard.getUpdatedAt() != null) {
                    SET("updated_at=#{storeBalanceCard.updatedAt}");
                }
                WHERE("store_id = #{storeBalanceCard.storeId}");
            } }.toString() + " limit " + limit;
        }
    
        public static String deleteById(int id, int limit) {
            return new SQL(){ {
                DELETE_FROM(STORE_BALANCE_CARDS);
                WHERE("store_balance_cards.id = #{id}");
            } }.toString() + " limit " + limit;
        }
    }
    
  6. 分页组件详解

十、接口输出

  1. 接口统一输出格式

    state: // 状态位
    msg: // 接口输出提示信息
    data: { // 总的接口输出数据,可以为空
        data: // 接口数据
        other: // 其他返回数据,如total/page_total等
        ...:
    }
    
  2. 正常接口输出/异常监听输出统一使用OutPut/ApiReturn方法

十一、异常处理

  1. 项目运行异常统一监听
    现有:ControllrHandler 路由异常监听、SqlHandler 数据库操作异常监听
    需要持续添加
  2. 统一使用接口输出类进行返回,杜绝直接返回报错信息。

十二、单元测试

  1. 添加单元测试类

    src\test\java\com\duomai\balance_card\BalanceCardApplicationTests.java

  2. demo

    @RunWith(SpringRunner.class)
    @SpringBootTest
    @AutoConfigureMockMvc
    public class BalanceCardApplicationTests {
    
        @Autowired
        private MockMvc mockMvc;
    
        @Test
        @SuppressWarnings("unchecked")
        public void getCard() throws Exception {
            String card_id = "1";
            String fields = "store_id,item_id,recharge_discount,photo,type,name";
            String res = this.mockMvc.perform(get("/card/get")
                    .param("id", card_id).param("fields", fields)
                    )
                    .andDo(print())
                    .andExpect(status().isOk())
                    .andReturn()
                    .getResponse()
                    .getContentAsString();
           // 接口返回不为空
           assertThat(res).isNotNull();
           // 校验接口返回格式是否完整
           Map<String, Object> api_res = Common.jsonToMap(res);
           assertThat(api_res).isNotNull();
           assertThat(api_res).hasSize(3);
           assertThat(api_res).containsKeys("state", "msg", "data");
           // 校验接口返回state是否正确
           int state = (int) api_res.get("state");
           Map<String, Object> api_data = (HashMap<String, Object>) api_res.get("data");
           assertThat(state).isEqualTo(ApiReturnDefines.SUCCESS);
           // 校验data数据是否正确
           assertThat(api_data).containsOnlyKeys("data");
           ArrayList<Map> data = (ArrayList) api_data.get("data");
           String[] fieldsAll = fields.split(",");
           assertThat(data).hasSize(1);
           assertThat(data.get(0)).containsKeys((Object[]) fieldsAll);
        }
    }
    
  3. 详解
    • 使用 @RunWith(SpringRunner.class)@SpringBootTest 定义测试类
    • 添加注解 @AutoConfigureMockMvc ,使用 Spring MockMvc 模拟Spring的HTTP请求并将其交给控制器,实际上并没有真正地启动服务器,仅仅是Mock,节省了启动服务器的开销。
    • 使用 AssertJ 库类来验证接口返回内容
      在demo中,使用断言验证了接口返回格式、state状态、data数据格式等基本内容。

    参考文档:https://spring.io/guides/gs/testing-web/

十三、项目打包

  1. 不同的构建文件
  • 普通 jar 包 : 会将源码编译后以工具包(即将class打成jar包)的形式对外提供,此时,你的 jar 包不一定要是可执行的,只要能通过编译,可以被别的项目以 import 的方式调用。
  • 可执行 jar 包 : 能通过 java -jar 的命令运行。
  • 普通 war 包 : war 是一个 web 模块,其中包括 WEB-INF,是可以直接运行的 WEB 模块。做好一个 web 应用后,打成包部署到容器中。
    • 可执行 war 包 : 普通 war 包 + 内嵌容器 。
  1. 构建可执行 jar 包

    • IDE打包
      右侧Gradle -> Tasks -> build -> build
    • 命令行打包
      进入项目根目录,执行gradle build
  2. 运行可执行 jar 包
    java -jar build/libs/balance_card-0.0.1.jar
    运行时可带参数,同application.properties中的参数名
    例:

    java -Djava.security.egd=file:/dev/./urandom -jar /opt/ci123/www/html/java/balance_card/build/libs/${jarName}.jar --server.port=80
    

十四、容器化部署

  1. 创建docker镜像
  • Dockfile 编写

    FROM java:8
    MAINTAINER duomai
    # 设置镜像源
    COPY sources.list /etc/apt/sources.list
    # 安装扩展
    RUN apt-get update && apt-get install -y \
    wget \
    curl \
    vim \
    git \
    less
    # 配置系统时间
    RUN /bin/cp /usr/share/zoneinfo/Asia/Shanghai /etc/localtime \
    && echo 'Asia/Shanghai' > /etc/timezone
    VOLUME /tmp
    # 设置工作目录
    ADD . /opt/ci123/www/html/java
    WORKDIR /opt/ci123/www/html/java
    # 指定输出端口
    EXPOSE 80

    • 快速构建镜像

      sh docker/build.sh [version]

  1. 环境变量配置
  • 配置

    #数据库连接配置 application.properties
    spring.datasource.url=jdbc:mysql://${APP_DB_HOST:192.168.0.235}:${APP_DB_PORT:3355}/${APP_DB_DATABASE:shop_balances}
    spring.datasource.username=${APP_DB_USER:cishop}
    spring.datasource.password=${APP_DB_PASSWORD:fuyuan1906}

    1. 启动容器( docker/run.sh )
  • 启动参数

    -p // 宿主机运行端口
    -d // 宿主机项目地址,缺省为执行run.sh的上一级目录
    -n // 容器名称
    -v // 镜像版本号
    -j // jar包版本,缺省为1.0

  • 环境变量

    # 运行配置
    APP_SERVER_PORT // 容器运行jar包的端口
    # 数据库配置
    APP_DB_HOST
    APP_DB_PORT
    APP_DB_DATABASE
    APP_DB_USER
    APP_DB_PASSWORD
    # 日志配置
    APP_LOG_PATH // 日志输出地址
    APP_LOG_SPRING_WEB_LEVEL // 对应logging.level.org.springframework.web,指org.springframework.web这个包下的日志输出等级,默认为ERROR,开发环境可配置为DEBUG
    APP_LOG_MYBATIS_LEVEL // 对应logging.level.com.duomai.balance_card.Model.Mapper,指balance_card.Model.Mapper包下的日志输出等级,默认为ERROR,开发环境可配置为DEBUG,开启DEBUG之后可在日志中查看DB操作

  • 启动容器

    docker run -d \
    -e APP_SERVER_PORT=8080 \
    -e APP_DB_HOST=192.168.0.235 \
    -e APP_DB_PORT=3355 \
    -e APP_DB_DATABASE=shop_balances \
    -e APP_DB_USER=cishop \
    -e APP_DB_PASSWORD=fuyuan1906 \
    -e APP_LOG_PATH=/opt/ci123/www/html/java/balance_card/log \
    -e APP_LOG_SPRING_WEB_LEVEL=DEBUG \
    -e APP_LOG_MYBATIS_LEVEL=DEBUG \
    -p $port:80 \
    -v $dir:/opt/ci123/www/html/java/balance_card \
    --restart=always \
    --name $name \
    harbor.oneitfarm.com/duomai/java-balance_card:$version \
    sh /opt/ci123/www/html/java/balance_card/docker/start.sh -j $jarName

通过mq消费者ip查找对应启动容器

  1. 首先先确定是否相应的消费者
  2. 3.14上有多个容器,如何确定是哪个容器运行的呢?
    • 由于在linux的docker中,容器里运行的进程会直接通过宿主机执行,所以可以通过查看是否有对应进程
      bash
      ps aux | grep cront/trades/Cashier_trade/payed

    • 获取对应进程的父级pid(也就是运行该脚本的pm2进程),比如这里选择第一个进程的pid为2163,获取到上级pid为29217
      bash
      # ps -ef|awk '$2 ~ /pid/{print $3}'
      ps -ef|awk '$2 ~ /2163/{print $3}'

    • 再获取pm2的上级(即运行容器的pid),获取到为24450
      bash
      ps -ef|awk '$2 ~ /29217/{print $3}'
    • 查看容器pid相对应的进程
      bash
      ps aux | grep 24450

    • 有一串容器编号,复制前一小段,查询对应容器
      bash
      docker ps | grep 0d7eccf5ab

【初阶】基于redis的抢购秒杀设计

秒杀系统架构思路

应用场景

限时抢购,10w人抢购100个商品。这种情况下如果直接走到数据存储层,瞬间会把数据库打挂掉,如果业务处理不好(流程过长),会出现超卖现象

优化方向

尽量将请求拦截在上游

充分利用缓存
1. 前端方案
- 页面静态
- 防重复提交
- 用户限流
2. 后端方案
- 网关限制
- 缓存
- 消息队列削峰

优化细节

这里我们由简单往细节深入(因为要考虑实际的用户量,先做最简单最有效的处理)

简单高效处理方式

  1. 前端防重复提交
    这是最基本的
  2. redis缓存校验
    如果没有缓存,由于业务校验时间过长,比如在200ms内10w个人同时读了数据库库存仅剩1,都是可购买的,然后同时走到下单流程就会超卖,另外就是如果并发量过高,数据库会挂掉。
    如果是缓存校验,则只有一个可下单。其余全部直接拦截。

redis数据类型选型
如果不考虑抢购的下单数量和购买限制大于1,那直接kv或者list都是可以的。否则可以考虑hash或者zset。

这里对比一下kv和list:
kv的一般处理方式是预加或者预减:累计购买数量超过总库存或者剩余库存小于0,则校验不通过,同时回退。
list:预存库存长度的list,不断pop,无法pop则校验不通过。

tip:注意原子性
判断最好走lua。
比如kv,如果库存1,10人同时下单,如果是程序判断,则全部不能下单。而如果用lua,用户判断和回退是原子性的,则有一个人可以下单成功。
如果使用list,当抢购数量大于1时,回退也需要用事务,不然会出现,比如库存3,A下单5,B下单2。B在回退push的过程中,又被Apop了1从而判断库存不足。导致两人都不能下单。

复杂完善处理方式

  1. redis集群,主从同步,读写分离(读多写少)。
  2. nginx负载均衡
  3. 前端资源静态化:只有少部分内容是动态的的
  4. 按钮控制
  5. 缓存预热
  6. mq削峰:队列下单

详细内容可以参考大佬们的文章,我只是一个搬运工

架构设计图:
秒杀架构设计

参考文档:
https://www.zhihu.com/question/54895548
https://yq.aliyun.com/articles/69704?utm_campaign=wenzhang&utm_medium=article&utm_source=QQ-qun&utm_content=m_10737

使用HAProxy实现RabbitMq集群负载均衡

一、目的

​ 在使用 RabbitMq集群时可能会遇到集群中某个节点出现异常或者连接数过多的情况,这个时候与该节点连接的Consumer将会断开,Publisher也会无法将消息发送至集群。为了解决这些问题,本文中将使用 HAProxy来代理集群,实现多个节点的负载均衡以及在某个节点异常时自动将连接切换至其他正常节点等功能。

二、HAProxy安装配置(Centos 7)

  1. 安装 HAProxy
    1. 下载最新稳定版2.0.8并解压
      // PWD:/opt/ci123/www/html/rabbitMq/
      wget https://www.haproxy.org/download/2.0/src/haproxy-2.0.8.tar.gz
      tar xf haproxy-2.0.8.tar.gz
      
    2. 查看系统内核版本来指定编译版本
      uname -r
      3.10.0-862.6.3.el7.x86_64
      

      版本参考:

      ```

    3. </ol>

      <ul>
      <li>linux22 for Linux 2.2</li>
      <li>linux24 for Linux 2.4 and above (default)</li>
      <li>linux24e for Linux 2.4 with support for a working epoll (> 0.21)</li>
      <li>linux26 for Linux 2.6 and above</li>
      <li>linux2628 for Linux 2.6.28, 3.x, and above (enables splice and tproxy)</li>
      <li>solaris for Solaris 8 or 10 (others untested)</li>
      <li>freebsd for FreeBSD 5 to 10 (others untested)</li>
      <li>netbsd for NetBSD</li>
      <li>osx for Mac OS/X</li>
      <li>openbsd for OpenBSD 5.7 and above</li>
      <li>aix51 for AIX 5.1</li>
      <li>aix52 for AIX 5.2</li>
      <li>cygwin for Cygwin</li>
      <li>haiku for Haiku</li>
      <li>generic for any other OS or version.</li>
      <li><p>custom to manually adjust every setting
      ```

      根据版本参考,这里我们选择linux2628版本进行编译。

      1. 编译到指定目录

        cd haproxy-2.0.8
        make TARGET=linux2628 PREFIX=/opt/ci123/haproxy
        
        // 这里出现报错,从2.0版本开始linux2628已被废弃
        Target 'linux2628' was removed from HAProxy 2.0 due to being irrelevant and
        often wrong. Please use 'linux-glibc' instead or define your custom target
        by checking available options using 'make help TARGET=<your-target>'.
        
        // 根据提示修改参数后编译
        make TARGET=linux-glibc PREFIX=/opt/ci123/haproxy
        make install PREFIX=/opt/ci123/haproxy
        
    4. 配置
      1. 复制 haproxy命令至全局变量
        cp /opt/ci123/haproxy/sbin/haproxy /usr/bin/
        
      2. 创建系统用户
        useradd -r haproxy
        
      3. 添加haproxy配置文件
        1. haproxy配置文件由五部分组成:
        • global: 参数是进程级的,通常和操作系统相关。这些参数一般只设置一次,如果配置无误,就不需要再次配置进行修改。
      • default:默认参数。
        • frontend:用于接收客户端请求的前端节点,可以设置相应的转发规则来指定使用哪个backend
        • backend:后端服务器代理配置,可实现代理多台服务器实现负载均衡、为请求添加额外报文数据等功能。
        • listen:是frontendbackend的结合,通常只对tcp流量有用。
      1. 添加配置文件/opt/ci123/haproxy/conf/haproxy.cfg

        ```
        # 全局配置
        global
        log 127.0.0.1 local3 # 设置日志
        pidfile /opt/ci123/haproxy/logs/haproxy.pid
        maxconn 4000 # 最大连接数
        user haproxy
        group haproxy
        daemon # 守护进程运行

        # 默认配置
        defaults
        log global
        mode tcp # 默认的模式mode { tcp|http|health },tcp是4层,http是7层,health只会返回OK
        option httplog # http 日志格式,仅在http模式下可用
        option dontlognull # 不记录健康检查日志信息;
        option redispatch # serverId对应的服务器挂掉后,强制定向到其他健康的服务器
        option http-server-close
        #option abortonclose # 当服务器负载很高的时候,自动结束掉当前队列处理比较久的链接;
        #option forwardfor # 如果后端服务器需要获得客户端真实ip需要配置的参数,可以从Http Header中获得客户端ip;
        #option httpclose # 主动关闭http通道,每次请求完毕后主动关闭http通道,ha-proxy不支持keep-alive,只能模拟这种模式的实现;<br />
        balance roundrobin # 负载均衡算法,轮询;
        retries 3 # 重试次数;

        <pre class="prism-highlight line-numbers" data-start="1"><code class="language-null"> timeout http-request 10s # 客户端建立连接但不请求数据时,关闭客户端连接;
        timeout queue 1m # 高负载响应haproxy时,会把haproxy发送来的请求放进一个队列中,timeout queue定义放入这个队列的超时时间;
        timeout connect 10s # 定义haproxy将客户端请求转发至后端服务器所等待的超时时间;
        timeout client 1m # 客户端非活动状态的超长时间(默认毫秒)
        timeout server 1m # 服务端与客户端非活动状态连接的超时时间。(默认毫秒)
        timeout http-keep-alive 10s # 定义保持连接的超时时长;
        timeout check 10s # 心跳检测超时;
        maxconn 3000 # 每个server最大的连接数;
        </code></pre>

        #前端配置
        frontend rabbitmq_cluster_front
        bind 0.0.0.0:10000 # http请求的端口,会被转发到设置的ip及端口
        default_backend rabbitmq_cluster_back

        # 后端配置
        backend rabbitmq_cluster_back
        #roundrobin 轮询方式
        balance roundrobin # 负载均衡的方式,轮询方式

        <pre class="prism-highlight line-numbers" data-start="1"><code class="language-null"> # 配置Rabbitmq连接负载均衡
        # 需要转发的ip及端口
        # inter 2000 健康检查时间间隔2秒
        # rise 3 检测多少次才认为是正常的
        # fall 3 失败多少次才认为是不可用的
        # weight 30 权重
        server clusterRabbit1 192.168.3.14:5672 check inter 2000 rise 3 fall 3 weight 30
        server clusterRabbit2 192.168.3.14:5673 check inter 2000 rise 3 fall 3 weight 30
        </code></pre>

        # 统计页面配置
        listen admin_stats<br />
        bind 0.0.0.0:10080 # 监听IP和端口,为了安全可以设置本机的局域网IP及端口;
        mode http
        option httplog # 采用http日志格式<br />
        stats refresh 30s # 统计页面自动刷新时间<br />
        stats uri /haproxy # 状态管理页面,通过/haproxy来访问
        stats realm Haproxy Manager # 统计页面密码框上提示文本<br />
        stats auth duomai:shijiemori@2012 # 统计页面用户名和密码设置<br />
        #stats hide-version # 隐藏统计页面上HAProxy的版本信息
        ```

        1. 配置日志 rsyslog
        vim /etc/rsyslog.conf
        # 取消如下2行注释
        $ModLoad imudp
        $UDPServerRun 51
        
        # 新增配置(自定义的日志设备)
        local3.*  /opt/ci123/haproxy/logs/haproxy.log
        
        # 重启rsyslog服务
        systemctl restart rsyslog
        
        1. 启动 haproxy
        haproxy -f /opt/ci123/haproxy/conf/haproxy.cfg
        

        访问统计页面出现如下界面:

        haproxy管理

    三、集群测试

    沿用上一篇【RabbitMq 镜像队列集群搭建】中的集群测试环境,在测试中将PubliserConsumer的连接替换为HAProxy的地址192.168.3.14:10000

    1. 测试环境
      1. 节点:
        1. 节点一:
        • clusterRabbit1
        • 端口:192.168.3.14:5672
          1. 节点二:
        • clusterRabbit2
        • 端口:192.168.3.14:5673
      2. 队列:
        1. 节点一 rabbit@clusterRabbit1
          1. 队列一:
          • nameclusterRabbit1Queue1
          • routing_keyclusterRabbit1key
          1. 队列二:
          • nameclusterRabbit1Queue2
          • routing_keyclusterRabbitCommonKey
        2. 节点二 rabbit@clusterRabbit2
          1. 队列三:
          • nameclusterRabbit2Queue1
          • routing_keyclusterRabbit2key
          1. 队列四:
          • nameclusterRabbit2Queue2
          • routing_keyclusterRabbitCommonKey(与队列二 相同)
    2. 启动消费者
      1. 消费者
        1. 消费者一:
        • 连接节点:节点一 rabbit@clusterRabbit1
        • 消费队列:队列一 clusterRabbit1Queue1
          1. 消费者二:
        • 连接节点:节点一 rabbit@clusterRabbit1
        • 消费队列:队列二 clusterRabbit1Queue2
          1. 消费者三:
        • 连接节点:节点二 rabbit@clusterRabbit2
        • 消费队列:队列三 clusterRabbit2Queue1
          1. 消费者四:
        • 连接节点:节点二 rabbit@clusterRabbit2
        • 消费队列:队列四 clusterRabbit2Queue2
      2. 启动结果:

        启动成功,但是在一分钟后客户端异常退出,原因是HAProxy设置了timeout client 1m 和 timeout server 1m,消费者在一分钟内都没有接收到消息导致被判定为不活跃连接从而被删除。

        由于HAProxy默认不支持长连接,上述问题可以使用pm2管理消费者的方法来解决,消费者进程在不活跃退出后pm2将自动重启此进程。

    3. 发布消息

      1. 发布clusterRabbit1key消息
      • 消费者一 成功收到消息。
        1. 发布clusterRabbitCommonKey消息
      • 消费者二/四 成功收到消息。
    4. 关闭节点二rabbit@clusterRabbit2后再次发布消息

    • 连接节点二的消费者先退出后重新使用HAProxy成功连接。
    • 发布的消息均能被成功消费。

    四、使用说明

    1. HAProxy默认不支持tcp 长连接,需要使用PM2之类的守护进程管理工具或者长连接技术来实现rabbitMq客户端持续连接。
    2. HAProxy通过活跃检测机制来判定负载均衡中的节点是否可用, 当rabbitMq集群中某个节点不可用时,在经过一段时间的活跃检测之后,HAProxy将弃用该节点直至节点恢复。在这种情况下,rabbitMq消费者将断开连接后选择剩余可用的节点再次启动,客户端发布时也会自动选择剩余可用的节点。
    3. rabbitMq集群中某个节点宕机之后,HAProxy会自动使用可用的集群节点,所以不会出现在HAProxy活跃检测期间发布消息出现一半成功一半失败的情况,所有的消息都将通过可用的集群节点发布至集群。

延迟队列使用说明

一、应用场景

  1. 对于商城系统而言,定时任务会非常多,比如优惠券过期、订单定时关闭、2小时订单未支付自动取消等等。常用做法是写数据库指定过期时间,定时循环读表,当数据量一大,这种会严重影响性能,产生巨大的io。
  2. 而另外一些基于redis实现的延迟队列,也是基于pull拉模式去实现的,也会产生io,只是变成了内存读取。
  3. 一种好的做法是 push 推模式,服务端主动推送,这里就是rabbitmq的死信队列去实现

二、延迟队列原理

延迟队列一般分为两种:
- 基于消息的延迟和基于队列的延迟。基于消息的延迟是指为每条消息设置不同的延迟时间,那么每当队列中有新消息进入的时候就会重新根据延迟时间排序,当然这也会对性能造成极大的影响。
- 实际应用中大多采用基于队列的延迟,设置不同延迟级别的队列,比如 5s、10s、30s、1min、5mins、10mins 等,每个队列中消息的延迟时间都是相同的,这样免去了延迟排序所要承受的性能之苦,通过一定的扫描策略(比如定时)即可投递超时的消息。
- 这里主要通过rabbitmq的死信队列实现,需要同时声明一个缓冲队列和一个延迟队列,给缓冲队列设置统一消息过期时间,当消息过期后,会自动被重新投递到死信队列(死信队列也是一个普通队列),只需要监听死信队列即可

三、api项目使用说明

  1. 首先需要定义延迟消费队列:即不管是订单关闭,还是优惠券过期的后续处理,都需要去监听这个消息
    在  application/libraries/mq/MqDlxQueue 去定义
    image.png
  2. 运行声明脚本
    php mq/StartDlx.php DELAY_QUEUE_TEST
  3. 定义缓冲区队列:即所有投递到这个队列的消息,设置统一的过期时间,比如固定2小时关闭订单,可以设置统一的队列过期时间为2小时。

    这里需要指定 dlx_key:即过期后投递到的死信队列的路由
    需要指定 message_ttl:即统一的消息过期时间,单位 毫秒

    注意:该 缓冲队列的 key,不要有任何消费者,不然被其他消费者消费后,就不会过期

    在 application/libraries/mq/MqQueue 去定义
    image.png

  4. 声明 缓冲队列
    php mq/Start.php BUFFER_QUEUE_TEST
  5. 设置 延迟队列消费者 
    image.png
  6. 发送 消息到 缓冲区
    image.png
  7. 测试效果:
    image.png

    image.png

RabbitMq 镜像队列集群搭建

建议食用本文前请先阅读【RabbitMq 普通集群搭建】

一、镜像队列集群概念

​ 镜像队列是基于普通集群模式的扩展,普通集群模式下如果某一个节点宕机,该节点下的队列操作将完全失效。而在镜像队列模式下,队列的数据将被复制到所有节点(或者配置过的节点)中,从而保证了一个节点宕机,其余节点也可以正常消费此消息。但此模式下也必然会带来性能下降、内存/磁盘消耗增加、网络IO负担增加等问题,所以镜像队列适用于对高可用要求比较高的系统。

  1. 每个镜像队列由一个主队列和一个或多个镜像组成。每个镜像队列都有自己的主节点,主队列存放于主节点上。对队列产生的操作将首先应用于队列的主节点,然后传播到镜像节点。包括发布队列、向消费者传递消息、跟踪来自消费者的确认等行为。
  2. 发布到集群中的消息将被复制到所有的镜像队列中,消费者连接任意节点消费队列实际上都将被连接至主队列的节点上。如果主队列已经确认消费了消息,则其余镜像队列中的消息将被丢弃。
  3. 如果主队列所在的节点发生异常,默认情况下最“老”的镜像队列将被选举为主队列,当然也可以制定不同的选举策略。

二、配置镜像队列

  1. 将队列配置成镜像队列需要通过创建policy来实现。policy包含策略键ha-mode和其对应的键值ha-params(可选)组成。
  • exactly模式
    • ha-paramscount,表示队列的总数量。
    • count表示主队列+镜像队列的总数量,如果count为1,则表示只存在于主队列。如果 count为2则表示存在主队列和一个镜像队列,以此类推。如果count值大于集群中节点的总数则表示所有节点都将同步一份镜像队列。如果某个镜像队列的节点宕机,则会寻找一个剩余未同步的节点来同步镜像队列。
  • all模式
    • 不需要ha-params
    • 此模式下所有节点都将同步镜像队列,如果有新节点加入,则新节点也会进行同步。官方建议同步镜像队列的节点数为N/2 + 1,其中N表示节点总数。同步到所有节点会增加所有集群节点的负载,包括网络I/O、磁盘I/O和磁盘空间的使用等。
  • nodes模式
    • ha-paramsnode names,节点的名称。
    • 此模式下将在指定的节点上同步镜像队列,如果声明队列时其余节点均不在新,则只会在声明连接的那个节点上创建队列。
  1. 配置policy
  • rabbitmqctl命令配置

    set_policy [-p vhost] [--priority priority] [--apply-to apply-to] name pattern definition

    • name:策略名称
    • pattern:策略匹配符,正则表达式。当与给定资源匹配时,将应用该策略。
    • definition:策略内容定义,JSON字符串。
    • priority:策略的优先级,整数。数字越大,优先级越高。默认值为0。
    • apply-to:策略应用的对象,支持queues,exchanges,all,默认值为all
      // exactly模式,匹配前缀为two的资源
      rabbitmqctl set_policy -p cluster ha-two ^two. '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'
      
      // all模式,匹配所有前缀
      rabbitmqctl set_policy -p cluster ha-all ^ '{"ha-mode":"all"}'
      
      // nodes模式,匹配所有前缀
      rabbitmqctl set_policy -p cluster ha-nodes ^ '{"ha-mode":"nodes","ha-params":["rabbit@nodeA", "rabbit@nodeB"]}'
      
  • management管理后台配置

    Admin -> Policies -> Add / update a policy

  1. 新镜像队列同步设置
  • ha-sync-mode:manual

    默认模式,新队列镜像将不同步现有消息,只接收新消息。当消费者消费完所有仅存在于主队列上的消息后,新的队列镜像将随着时间的推移成为和主节点队列相同的精确镜像队列。

  • ha-sync-mode: automatic

    当新队列镜像加入时,队列将自动同步所有消息。

三、额外说明

  1. 独占队列不会被复制为镜像队列。
  2. 主节点队列失效后,其中一个镜像队列将被选举为主队列并带来如下影响:
    1. 与主机点连接的客户端将全部断开。
    2. 运行时间最长的镜像队列将被选举为主队列,如果镜像队列尚未开始同步,则队列上的消息将丢失。
    3. 新的主队列会认为之前所有的消费者的连接都已经断开,它将重新发送旧队列中没有收到ack的消息。这可能出现客户端已经发送过ack, 但是服务端在接收到之前就已宕机的情况,从而导致发送两遍相同的消息,因此所有未确认的消息都将使用redelivered标志重新发送。
    4. 如果消费者连着的是镜像队列节点,并且消费者在启动时设置了x-cancel-on-ha-failover参数,则消费者将收到一个服务端消费取消的通知,如果未设置此参数,则消费者将无法感知主节点已宕机。
    5. 如果使用自动ack机制则消息将丢失。
  3. 如果停止了某个包含主节点队列的节点,则其他节点的镜像队列将被选举为主节点队列。在重新启动此节点后,该节点只会被当做是一个新加入集群的节点,不会重新成为主节点队列。
  4. 在主节点队列宕机并且其他镜像队列尚未同步的极端情况下,rabbitMq集群将拒绝任何镜像队列选举为主队列,整个队将不可用且被关闭。如果在镜像队列尚未同步的情况下也需要将某个镜像队列选举为主队列,需要配置policyha-promote-on-shutdownalways(默认为when-synced),并且ha-promote-on-failure不可配置为hen-synced(默认值为always)。

四、集群测试

  1. 启动节点、创建用户、vhostexchangequeue,启动消费者。(快速启动,参考【RabbitMq 使用docker搭建集群】篇)

    1. 节点:
      1. 节点一:rabbit@clusterRabbit1
      2. 节点二:rabbit@clusterRabbit2
    2. 用户:api_managementadministrator标签,开放虚拟主机cluster所有权限)

    3. 策略:

      rabbitmqctl set_policy -p cluster ha-all ^ '{"ha-mode":"all"}'
      
    4. vhostcluster

    5. exchangecluster(直连交换机)

    6. queue

      1. 节点一 rabbit@clusterRabbit1
        1. 队列一:
        • nameclusterRabbit1Queue1
        • routing_keyclusterRabbit1key
        1. 队列二:
        • nameclusterRabbit1Queue2
        • routing_keyclusterRabbitCommonKey
      2. 节点二 rabbit@clusterRabbit2
        1. 队列三:
        • nameclusterRabbit2Queue1
        • routing_keyclusterRabbit2key
        1. 队列四:
        • nameclusterRabbit2Queue2
        • routing_keyclusterRabbitCommonKey(与队列二 相同)
    7. consumer
      1. 消费者一:
      • 连接节点:节点一 rabbit@clusterRabbit1
      • 消费队列:队列一 clusterRabbit1Queue1
        1. 消费者二:
      • 连接节点:节点一 rabbit@clusterRabbit1
      • 消费队列:队列二 clusterRabbit1Queue2
        1. 消费者三:
      • 连接节点:节点二 rabbit@clusterRabbit2
      • 消费队列:队列三 clusterRabbit2Queue1
        1. 消费者四(非此节点的队列):
      • 连接节点:节点一 rabbit@clusterRabbit1
      • 消费队列:队列三 clusterRabbit2Queue1
        1. 消费者五:
      • 连接节点:节点二 rabbit@clusterRabbit2
      • 消费队列:队列四 clusterRabbit2Queue2
  2. 连接节点二 rabbit@clusterRabbit2,再次声明队列名和队列一同名的队列clusterRabbit1Queue1。【同名队列再次声明】

  • 结果:声明不成功,节点二中没有生成新的clusterRabbit1Queue1队列,而节点一中clusterRabbit1Queue1队列多出了新绑定的routing_key:clusterRabbit2key,这意味着在同一集群中不同节点之间的队列名是唯一的,在一个节点中可以操作另一个节点的队列数据。
  1. 生产者连接节点一 rabbit@clusterRabbit1,发布clusterRabbit1key消息。【发布此节点队列消息】
  • 消费者一 成功接收到消息。
  1. 生产者连接节点二 rabbit@clusterRabbi2,发布clusterRabbit1key消息。【发布非此节点队列消息】
  • 消费者一 成功接收到消息。
  1. 生产者连接节点一 rabbit@clusterRabbit1,发布clusterRabbitCommonKey消息。【多个节点队列绑定相同消息】
  • 消费者二 成功接收到消息。
  • 消费者五 成功接收到消息。
  1. 生产者连接节点一 rabbit@clusterRabbit1,发布clusterRabbit2key消息。【消费非此节点队列消息】
  • 消费三、四 轮询接收到消息。
  1. 关闭节点二 rabbit@clusterRabbit2,连接节点一 rabbit@clusterRabbit1,发布clusterRabbit2key消息【投递消息给集群中意外退出的节点】
  • 连接节点二 rabbit@clusterRabbit2的消费者全部断开,消费者四并未断开。
  • 节点一rabbit@clusterRabbit1 晋升成为队列三、队列四的主节点。
  • 消费者四 可以继续成功接收到消息。
  • 再次启动节点二rabbit@clusterRabbit2,没有再次成为队列三、队列四的主节点,只是复制了队列三、队列四的镜像队列。
  1. 修改消费者四,在ack之前sleep(20),发布clusterRabbit2key消息,并立即关闭节点二 rabbit@clusterRabbit2
  • 和上面一样,消费者四并未断开。
  • 消费者收到两条相同的消息,说明节点二在收到 ack之前宕机后,此消息会被当做未消费的消息重新放入队列后消费。

RabbitMq 使用docker搭建集群

建议食用本文前请先阅读【RabbitMq 普通集群搭建】

一、RabbitMq镜像

  1. 镜像: rabbitmq:3.8-management
  2. 启动参数:
    • --name: 容器名称
    • -h / --hostnamerabbitMq默认节点的 host
    • -v:文件挂载映射
    • -p:端口映射
      • 5672:容器内默认的rabbitMq 启动端口
      • 15672:容器内默认的rabbitMq 管理插件启动端口
  3. 环境变量:(支持所有rabbitMq环境变量)
    • RABBITMQ_NODENAME:节点名称,缺省为 rabbit@[hostname]
    • RABBITMQ_DEFAULT_USER:默认用户名,缺省为guset
    • RABBITMQ_DEFAULT_PASS:默认密码,缺省为 guest
    • RABBITMQ_DEFAULT_VHOST:默认虚拟主机,缺省为/
    • RABBITMQ_ERLANG_COOKIEerlang.cookie

二、搭建集群

  1. 启动多份rabbitMq容器
    # 启动节点一
    docker run -d --hostname clusterRabbit1 --name clusterRabbit1 -p 15672:15672 -p 5672:5672 -e RABBITMQ_ERLANG_COOKIE='rabbitcookie' -v /opt/ci123/www/html/rabbitMq/clusterRabbit1:/var/lib/rabbitmq rabbitmq:3.8-management
    
    # 启动节点二
    docker run -d --hostname clusterRabbit2 --name clusterRabbit2 -p 5673:5672 --link clusterRabbit1:clusterRabbit1 -e RABBITMQ_ERLANG_COOKIE='rabbitcookie' -v /opt/ci123/www/html/rabbitMq/clusterRabbit2:/var/lib/rabbitmq rabbitmq:3.8-management
    
  • /var/lib/rabbitmq 是容器内部文件数据存放目录,使用-v进行文件目录挂载。
  • 多个容器之间使用--link <name or id>:alias连接,否则需要自行在各个容器添加供rabbitMq 相互访问的host
  • Erlang Cookie值必须相同,rabbitMQ是通过Erlang实现的,Erlang Cookie相当于不同节点之间相互通讯的秘钥,Erlang节点通过交换Erlang Cookie获得认证。
  1. 启动集群
    1. 设置节点二 rabbit@clusterRabbit2,加入节点一 rabbit@clusterRabbit1的集群
      rabbitmqctl stop_app
      rabbitmqctl reset
      rabbitmqctl join_cluster rabbit@clusterRabbit1
      rabbitmqctl start_app
      
    2. 查看rabbitMq管理后台集群数据

      rabbitmq管理后台

  2. 一键化启动脚本

    #!bin/sh
    # 启动节点一
    docker run -d --hostname clusterRabbit1 --name clusterRabbit1 -p 15672:15672 -p 5672:5672 -e RABBITMQ_ERLANG_COOKIE='rabbitcookie' -v /opt/ci123/www/html/rabbitMq/clusterRabbit1:/var/lib/rabbitmq rabbitmq:3.8-management
    
    # 启动节点二
    docker run -d --hostname clusterRabbit2 --name clusterRabbit2 -p 5673:5672 --link clusterRabbit1:clusterRabbit1 -e RABBITMQ_ERLANG_COOKIE='rabbitcookie' -v /opt/ci123/www/html/rabbitMq/clusterRabbit2:/var/lib/rabbitmq rabbitmq:3.8-management
    
    echo "即将开始初始化"
    sleep 5
    # 加入集群
    docker exec -it clusterRabbit2 sh -c 'rabbitmqctl stop_app && rabbitmqctl reset && rabbitmqctl join_cluster rabbit@clusterRabbit1 && rabbitmqctl start_app'
    
    # 添加虚拟主机
    docker exec -it clusterRabbit1 sh -c 'rabbitmqctl add_vhost cluster'
    
    # 添加用户
    docker exec -it clusterRabbit1 sh -c 'rabbitmqctl add_user api_management shijiemori2012'
    
    # 添加用户权限
    docker exec -it clusterRabbit1 sh -c 'rabbitmqctl set_permissions -p cluster api_management ".*" ".*" ".*"'
    
    # 设置用户标签
    docker exec -it clusterRabbit1 sh -c 'rabbitmqctl set_user_tags api_management administrator'
    

RabbitMq 普通集群搭建

一、集群概念

RabbitMQ集群是一个或多个节点的逻辑分组,每个节点共享用户、虚拟主机、队列、交换、绑定路由、运行时参数和其他分布式状态。集群中的节点可以动态地添加/删除,RabbitMQ代理一开始都运行在单个节点上,可以将这些节点连接到集群中,然后再将其转换回各个代理。

  1. 默认情况下,RabbitMq将复制除消息队列外的所有数据至集群中的每一个节点。而消息队列的完整数据只会存放于创建该队列的节点上,其余节点仅保存该队列的元数据和指针(类似于索引)。如果需要复制队列,则需要启用镜像队列集群。

  2. 集群中每个节点是平等的,不存在主从和特殊的节点。

  3. 集群中的节点通过Erlang Cookie相互通信,每个节点必须具有相同的cookie

  4. 节点分为磁盘节点和RAM节点,RAM节点只在RAM中存储内部数据库表,并不存储包括消息、消息存储索引、队列索引和其他节点状态等数据,RAM节点的性能更加高效,但是由于数据是非持久化的,一旦宕机将无法恢复。默认创建的都是磁盘节点。

  5. 单节点拓扑图如下,集群的拓扑是基于多个Node节点的扩展。

    单节点拓扑图

二、配置需求

  1. 配置方式
    • config 文件配置
    • rabbitmqctl命令配置(下文中使用此方法配置)
  2. 集群中的节点名必须是唯一的
    • 可以在启动时使用环境变量RABBITMQ_NODENAME设置
    • 节点名由[节点名称]@[host]组成
  3. 各个节点的启动端口可以被成功连接
  4. 节点之间通过节点名相互访问,要求各个节点之间的host可以相互进行DNS解析
  5. 每个节点之间必须配置相同的Erlang Cookie(多机环境需要额外配置)

三、集群配置

  1. 启动多个独立的节点

    RABBITMQ_NODE_PORT=5674 RABBITMQ_NODE_IPDDRESS=192.168.0.235  RABBITMQ_SERVER_START_ARGS="-rabbitmq_management listener [{port,15673}]" RABBITMQ_NODENAME="clusterRabbit1@VM235"  /usr/local/rabbitmq/3.1.3/sbin/rabbitmq-server -detached
    RABBITMQ_NODE_PORT=5675 RABBITMQ_NODE_IPDDRESS=192.168.0.235  RABBITMQ_SERVER_START_ARGS="-rabbitmq_management listener [{port,15674}]" RABBITMQ_NODENAME="clusterRabbit2@VM235" /usr/local/rabbitmq/3.1.3/sbin/rabbitmq-server -detached
    

    执行完成后,分别创建了名称为clusterRabbit1@VM235,clusterRabbit2@VM235的两个节点。

    VM235需要提前配置host文件,确保各个节点之间的host可以相互进行DNS解析。

  2. 创建集群

    1. 关闭节点clusterRabbit1@VM235
      ./rabbitmqctl -n clusterRabbit1@VM235 stop_app
      
    2. 重置节点clusterRabbit1@VM235
      ./rabbitmqctl -n clusterRabbit1@VM235 reset
      

      必须重置节点才能加入现有集群,重置节点将删除该节点上以前存在的所有资源和数据。这意味着节点不能在成为集群的成员时保留其现有数据,节点中的数据需要进行额外的备份和恢复。

    3. 将节点clusterRabbit1@VM235加入clusterRabbit2@VM235的集群

      ./rabbitmqctl -n clusterRabbit1@VM235 join_cluster clusterRabbit2@VM235
      
    4. 启动节点clusterRabbit1@VM235
      ./rabbitmqctl -n clusterRabbit1@VM235 start_app
      
    5. 查看集群信息
      ./rabbitmqctl -n clusterRabbit1@VM235 cluster_status
      结果:
      Cluster status of node clusterRabbit1@VM235 ...
      [{nodes,[{disc,[clusterRabbit1@VM235,clusterRabbit2@VM235]}]},
      {running_nodes,[clusterRabbit2@VM235,clusterRabbit1@VM235]},
      {partitions,[]}]
      ...done.
      
  3. 集群中节点的关闭与重启
    1. 节点关闭并重启之后会选择一个在线的集群成员(只考虑磁盘节点)进行同步。在重新启动节点时,默认情况下将尝试与该成员联系10次,并有30秒的响应超时。如果该成员在时间间隔内可用则节点将成功启动,并与该成员同步所需内容后继续运行。如果该成员无法响应,则重新启动的节点将放弃同步数据并启动。
    2. 以下情况将导致节点无法重新加入集群:
    • 修改节点名/主机名,节点的数据目录路径会因此更改。
    • 重置节点数据/更换节点数据目录
  4. 移除集群中的节点
    1. 关闭该节点
    2. 重置该节点
    3. 再次启动该节点

四、集群测试

  1. 创建用户、vhostexchangequeue,并启动消费者。【测试基础】
    1. 用户:api_managementmanagement标签,开放虚拟主机cluster所有权限)
    2. vhostcluster
    3. exchangecluster(直连交换机)
    4. queue
      1. 节点一 clusterRabbit1
        1. 队列一:
        • nameclusterRabbit1Queue1
        • routing_keyclusterRabbit1key
        1. 队列二:
        • nameclusterRabbit1Queue2
        • routing_keyclusterRabbitCommonKey
      2. 节点二 clusterRabbit2
        1. 队列三:
        • nameclusterRabbit2Queue1
        • routing_keyclusterRabbit2key
        1. 队列四:
        • nameclusterRabbit2Queue2
        • routing_keyclusterRabbitCommonKey(与队列二 相同)
    5. consumer
      1. 消费者一:
      • 连接节点:节点一 clusterRabbit1
      • 消费队列:队列一 clusterRabbit1Queue1
        1. 消费者二:
      • 连接节点:节点一 clusterRabbit1
      • 消费队列:队列二 clusterRabbit1Queue2
        1. 消费者三(非此节点的队列):
      • 连接节点:节点一 clusterRabbit1
      • 消费队列:队列三 clusterRabbit2Queue1
        1. 消费者四:
      • 连接节点:节点二 clusterRabbit2
      • 消费队列:队列四 clusterRabbit2Queue2
  2. 连接节点二 clusterRabbit2,再次声明队列名和队列一同名的队列clusterRabbit1Queue1。【同名队列再次声明】
    • 结果:声明不成功,节点二中没有生成新的clusterRabbit1Queue1队列,而节点一中clusterRabbit1Queue1队列多出了新绑定的routing_key,这意味着在同一集群中不同节点之间的队列名是唯一的,在一个节点中可以操作另一个节点的队列数据。
  3. 生产者连接节点一 clusterRabbit1,发布clusterRabbit1key消息。【发布此节点队列消息】
    • 消费者一 成功接收到消息
  4. 生产者连接节点二 clusterRabbit2,发布clusterRabbit1key消息。【发布非此节点队列消息】
    • 消费者一 成功接收到消息
  5. 生产者连接节点一 clusterRabbit1,发布clusterRabbitCommonKey消息。【多个节点队列绑定相同消息】
    • 消费者二 成功接收到消息
    • 消费者四 成功接收到消息
  6. 生产者连接节点一 clusterRabbit1,发布clusterRabbit2key消息。【消费非此节点队列消息】
    • 消费三 成功接收到消息
  7. 关闭节点二 clusterRabbit2,连接节点一,发布clusterRabbit2key消息【投递消息给集群中意外退出的节点】
    • 关闭节点二 clusterRabbit2之后,消费者三、四异常退出
    • 投递消息至集群成功
    • 再次启动节点二 clusterRabbit2以及消费者三、四,之前投递的消息没有成功接收

五、使用总结

  1. 同一集群中不同节点之间的队列名是唯一的,在一个节点中可以操作另一个节点的队列数据。
  2. 生产者可以发布集群中任一节点队列绑定的消息,集群将自动匹配出符合条件的节点队列,并投递给消费者进行消费。
  3. 集群中不同节点的队列如果绑定了相同的routing_key,消息将投递到集群中所有符合路由匹配条件的节点队列中。
  4. 消费者可以订阅集群中任意节点的队列。
  5. 集群中某个节点异常退出后,生产者投递到集群中的消息将无法送达至该节点,但是不影响其他节点的接收。(解决这个问题需要使用镜像队列集群)

六、错误记录

  1. 启动新的节点报错:could_not_start,rabbitmq_management

    解决:

    rabbitmq_management插件默认使用的是15672端口,这个端口已被之前启动的节点占用,修改启动命名为 rabbitmq_management插件指定一个新的端口即可。

    RABBITMQ_NODE_PORT=5674 RABBITMQ_NODE_IPDDRESS=192.168.0.235  RABBITMQ_SERVER_START_ARGS="-rabbitmq_management listener [{port,15673}]" RABBITMQ_NODENAME="clusterRabbit1@VM235"  /usr/local/rabbitmq/3.1.3/sbin/rabbitmq-server -detached
    
  2. 【认知错误】rabbitmq_management管理后台Overview中节点的Memory表示使用的内存,Disk space表示剩余可用的磁盘空间,不是已使用的磁盘空间。

  3. 修改节点的host后启动失败:``ERROR: epmd error for host "VM235": nxdomain (non-existing domain)`

    解决:

    VM235需要预先添加到 HOST文件,确保可以被正确解析。

  4. 节点一 clusterRabbit1 退出集群后重新加入集群,再次声明之前的队列提示routing_key绑定不成功

    NOT_FOUND - no binding clusterRabbitCommonKey between exchange 'cluster' in vhost 'cluster' and queue 'clusterRabbit1Queue2' in vhost 'cluster'
    

    此问题是rabbitMq集群本身的问题,且未得到官方明确的解决方案。

    一些临时解放方案:

redis实战经验

如何避免key冲突?

  • select db
    • 默认db16个,编号0~15,可通过配置修改db个数
    • 默认使用db-0
  • key命名格式
    • 以"xxx:yyy:zzz"格式命名

安全考虑

  • 访问限制
    • 只对内网访问,以防外部通过6379端口访问
  • 危险函数
    • 禁用危险函数:flush、flushall
  • 使用秘钥
    • $redis->auth(password)

性能提升

  • 连接
    • 使用官方扩展,弃用predis
    • 使用单例方式连接,使用pconnect
    • 用中间件,实现连接池
  • 持久化策略
    • 根据实际情况,关闭持久化
  • 服务部署
    • 与web服务共存,本地调用
    • 与存储服务隔离(mysql),避免高io
    • 一主多从,提升性能,或使用集群
  • 版本升级
    • 低版本:setnx + expire
    • 高版本:set(k, v, array(ex, nx))
  • 批量操作
    • 使用管道命令,批量导入数据
    • 使用mset, mget
  • 合理操作
    • 使用hset存放json,而不是set
    • 使用scan代替keys
    • Nginx + Lua + Redis

问题:
如何检查redis服务是否健康?