薛定谔的风口猪

站在巨人的肩膀上学习,猪都能看得很远

RocketMQ——角色与术语详解

RocketMQ中有很多概念,其中包括一些术语和角色。

理清楚基本的概念能有效的帮助理解RocketMQ的原理以及排查问题。

角色:

Producer

生产者。发送消息的客户端角色。发送消息的时候需要指定Topic。

Consumer

消费者。消费消息的客户端角色。通常是后台处理异步消费的系统。 RocketMQ中Consumer有两种实现:PushConsumer和PullConsumer。

PushConsumer

推送模式(虽然RocketMQ使用的是长轮询)的消费者。消息的能及时被消费。使用非常简单,内部已处理如线程池消费、流控、负载均衡、异常处理等等的各种场景。

PullConsumer

拉取模式的消费者。应用主动控制拉取的时机,怎么拉取,怎么消费等。主动权更高。但要自己处理各种场景。

概念术语

Producer Group

标识发送同一类消息的Producer,通常发送逻辑一致。发送普通消息的时候,仅标识使用,并无特别用处。若事务消息,如果某条发送某条消息的producer-A宕机,使得事务消息一直处于PREPARED状态并超时,则broker会回查同一个group的其 他producer,确认这条消息应该commit还是rollback。但开源版本并不支持事务消息。

Consumer Group

标识一类Consumer的集合名称,这类Consumer通常消费一类消息,且消费逻辑一致。同一个Consumer Group下的各个实例将共同消费topic的消息,起到负载均衡的作用。

消费进度以Consumer Group为粒度管理,不同Consumer Group之间消费进度彼此不受影响,即消息A被Consumer Group1消费过,也会再给Consumer Group2消费。

注: RocketMQ要求同一个Consumer Group的消费者必须要拥有相同的注册信息,即必须要听一样的topic(并且tag也一样)。

Topic

标识一类消息的逻辑名字,消息的逻辑管理单位。无论消息生产还是消费,都需要指定Topic。

Tag

RocketMQ支持给在发送的时候给topic打tag,同一个topic的消息虽然逻辑管理是一样的。但是消费topic1的时候,如果你订阅的时候指定的是tagA,那么tagB的消息将不会投递。

Message Queue

简称Queue或Q。消息物理管理单位。一个Topic将有若干个Q。若Topic同时创建在不通的Broker,则不同的broker上都有若干Q,消息将物理地存储落在不同Broker结点上,具有水平扩展的能力。

无论生产者还是消费者,实际的生产和消费都是针对Q级别。例如Producer发送消息的时候,会预先选择(默认轮询)好该Topic下面的某一条Q地发送;Consumer消费的时候也会负载均衡地分配若干个Q,只拉取对应Q的消息。

每一条message queue均对应一个文件,这个文件存储了实际消息的索引信息。并且即使文件被删除,也能通过实际纯粹的消息文件(commit log)恢复回来。

Offset

RocketMQ中,有很多offset的概念。但通常我们只关心暴露到客户端的offset。一般我们不特指的话,就是指逻辑Message Queue下面的offset。

注: 逻辑offset的概念在RocketMQ中字面意思实际上和真正的意思有一定差别,这点在设计上显得有点混乱。祥见下面的解释。

可以认为一条逻辑的message queue是无限长的数组。一条消息进来下标就会涨1,而这个数组的下标就是offset。

max offset

字面上可以理解为这是标识message queue中的max offset表示消息的最大offset。但是从源码上看,这个offset实际上是最新消息的offset+1,即:下一条消息的offset。

min offset:

标识现存在的最小offset。而由于消息存储一段时间后,消费会被物理地从磁盘删除,message queue的min offset也就对应增长。这意味着比min offset要小的那些消息已经不在broker上了,无法被消费。

consumer offset

字面上,可以理解为标记Consumer Group在一条逻辑Message Queue上,消息消费到哪里即消费进度。但从源码上看,这个数值是消费过的最新消费的消息offset+1,即实际上表示的是下次拉取的offset位置

消费者拉取消息的时候需要指定offset,broker不主动推送消息, offset的消息返回给客户端。

consumer刚启动的时候会获取持久化的consumer offset,用以决定从哪里开始消费,consumer以此发起第一次请求。

每次消息消费成功后,这个offset在会先更新到内存,而后定时持久化。在集群消费模式下,会同步持久化到broker,而在广播模式下,则会持久化到本地文件。

集群消费

消费者的一种消费模式。一个Consumer Group中的各个Consumer实例分摊去消费消息,即一条消息只会投递到一个Consumer Group下面的一个实例。

实际上,每个Consumer是平均分摊Message Queue的去做拉取消费。例如某个Topic有3条Q,其中一个Consumer Group 有 3 个实例(可能是 3 个进程,或者 3 台机器),那么每个实例只消费其中的1条Q。

而由Producer发送消息的时候是轮询所有的Q,所以消息会平均散落在不同的Q上,可以认为Q上的消息是平均的。那么实例也就平均地消费消息了。

这种模式下,消费进度的存储会持久化到Broker。

广播消费

消费者的一种消费模式。消息将对一个Consumer Group下的各个Consumer实例都投递一遍。即即使这些 Consumer 属于同一个Consumer Group,消息也会被Consumer Group 中的每个Consumer都消费一次。

实际上,是一个消费组下的每个消费者实例都获取到了topic下面的每个Message Queue去拉取消费。所以消息会投递到每个消费者实例。

这种模式下,消费进度会存储持久化到实例本地。

顺序消息

消费消息的顺序要同发送消息的顺序一致。由于Consumer消费消息的时候是针对Message Queue顺序拉取并开始消费,且一条Message Queue只会给一个消费者(集群模式下),所以能够保证同一个消费者实例对于Q上消息的消费是顺序地开始消费(不一定顺序消费完成,因为消费可能并行)。

在RocketMQ中,顺序消费主要指的是都是Queue级别的局部顺序。这一类消息为满足顺序性,必须Producer单线程顺序发送,且发送到同一个队列,这样Consumer就可以按照Producer发送的顺序去消费消息。

生产者发送的时候可以用MessageQueueSelector为某一批消息(通常是有相同的唯一标示id)选择同一个Queue,则这一批消息的消费将是顺序消息(并由同一个consumer完成消息)。或者Message Queue的数量只有1,但这样消费的实例只能有一个,多出来的实例都会空跑。

普通顺序消息

顺序消息的一种,正常情况下可以保证完全的顺序消息,但是一旦发生异常,Broker宕机或重启,由于队列总数发生发化,消费者会触发负载均衡,而默认地负载均衡算法采取哈希取模平均,这样负载均衡分配到定位的队列会发化,使得队列可能分配到别的实例上,则会短暂地出现消息顺序不一致。

如果业务能容忍在集群异常情况(如某个 Broker 宕机或者重启)下,消息短暂的乱序,使用普通顺序方式比较合适。

严格顺序消息

顺序消息的一种,无论正常异常情况都能保证顺序,但是牺牲了分布式 Failover 特性,即 Broker集群中只要有一台机器不可用,则整个集群都不可用,服务可用性大大降低。

如果服务器部署为同步双写模式,此缺陷可通过备机自动切换为主避免,不过仍然会存在几分钟的服务不可用。(依赖同步双写,主备自动切换,自动切换功能目前并未实现)

目前已知的应用只有数据库 binlog 同步强依赖严格顺序消息,其他应用绝大部分都可以容忍短暂乱序,推荐使用普通的顺序消息

RocketMQ——组件

rocketmq部署

RocketMQ服务端的组件有三个,NameServer,Broker,FilterServer(可选,部署于和Broker同一台机器)

下面分别介绍三个组件:

Name Server

Name Server是RocketMQ的寻址服务。用于把Broker的路由信息做聚合。用户端依靠Name Server决定去获取对应topic的路由信息,从而决定对哪些Broker做连接。

  • Name Server是一个几乎无状态的结点,Name Server之间采取share-nothing的设计,互不通信。

  • 对于一个Name Server集群列表,客户端连接Name Server的时候,只会选择随机连接一个结点,以做到负载均衡。

  • Name Server所有状态都从Broker上报而来,本身不存储任何状态,所有数据均在内存。

  • 如果中途所有Name Server全都挂了,影响到路由信息的更新,不会影响和Broker的通信。

Broker

Broker是处理消息存储,转发等处理的服务器。

  • Broker以group分开,每个group只允许一个master,若干个slave。
  • 只有master才能进行写入操作,slave不允许。
  • slave从master中同步数据。同步策略取决于master的配置,可以采用同步双写,异步复制两种。
  • 客户端消费可以从master和slave消费。在默认情况下,消费者都从master消费,在master挂后,客户端由于从Name Server中感知到Broker挂机,就会从slave消费。
  • Broker向所有的NameServer结点建立长连接,注册Topic信息。

Filter Server(可选)

RocketMQ可以允许消费者上传一个Java类给Filter Server进行过滤。

  • Filter Server只能起在Broker所在的机器
  • 可以有若干个Filter Server进程
  • 拉取消息的时候,消息先经过Filter Server,Filter Server靠上传的Java类过滤消息后才推给Consumer消费。
  • 客户端完全可以消费消息的时候做过滤,不需要Filter Server
  • FilterServer存在的目的是用Broker的CPU资源换取网卡资源。因为Broker的瓶颈往往在网卡,而且CPU资源很闲。在客户端过滤会导致无需使用的消息在占用网卡资源。
  • 使用 Java class上传作为过滤表达式是一个双刃剑——一方面方便了应用的过滤操作且节省网卡资源,另一方面也带来了服务器端的安全风险。这就需要应用来保证过滤代码安全——例如在过滤程序里尽可能不做申请大内存,创建线程等操作。避免 Broker 服务器资源泄漏。

在IntilliJ中运行Maven的Tomcat项目

对于一个Maven项目,如果又是一个tomcat项目,在运行tomcat之前应该要进行Maven的构建。

以下是步骤:

1.新建一个Maven Run/Debug configuration:

新建一个Maven Run/Debug configuration

2.在Tomcat Run/Debug Configuration中新增#1的阶段在”Before Launch”中

在Tomcat Run/Debug Configuration中新增#1的阶段在"Before Launch"中

如何优雅地停止运行中的内嵌Tomcat的Spring Boot应用

你很可能根据https://spring.io/guides/gs/rest-service/搭起了一个Spring的Rest服务,然后打包成了jar包,不需要容器就可以在生成环境下通过运行jar包启动一个Web服务。

但这样的服务怎么样正确的停止呢?或许你只是简单的kill -9对应的进程,但实际上,有更优雅的方式。

Spring Boot里面有一个spring-boot-starter-actuator的项目,可以监控和管理Spring Boot应用。其中暴露了很多endpoint,可以方便的检测应用的健康情况。其中有一个shutdown的endpoint可以优雅地停止应用。

通过VM参数指定本地log4j配置文件,Spring Boot

Spring Boot有预设的日志配置逻辑(具体参看:这里), 如果是log4j的话,以下文件会被加载:

log4j-spring.properties, log4j-spring.xml, log4j.properties , log4j.xml

有时候我们项目的log4j配置文件配置的是生产环境,每次本地调试又不想改会本地的调试配置,若希望通过VM参数去修改此文件,按照之前的参数-Dlog4j.configuration(非Spring Boot项目请看这里通过VM参数选择本地log4j配置文件),在Spring Boot的项目中并不生效。

若需要指定另外的文件,需要用Spring Boot指定的配置:-Dlogging.config

-Dlogging.config=D:\project\git_repo\prome-data\src\main\resources\log4j-debug.properties

即可在本地运行时选择本地的配置文件进行日志配置。

避免写出不走索引的SQL, MySQL

在MySQL中,并不是你建立了索引,并且你在SQL中使用到了该列,MySQL就肯定会使用到那些索引的,有一些情况很可能在你不知不觉中,你就“成功的避开了”MySQL的所有索引。

现假设有t_stu表,age,sname上建立了索引

索引列参与计算

如果where条件中age列中使用了计算,则不会使用该索引

SELECT `sname` FROM `t_stu` WHERE `age`=20;-- 会使用索引
SELECT `sname` FROM `t_stu` WHERE `age`+10=30;-- 不会使用索引!!因为所有索引列参与了计算
SELECT `sname` FROM `t_stu` WHERE `age`=30-10;-- 会使用索引

故,如果需要计算,千万不要计算到索引列,想方设法让其计算到表达式的另一边去。

索引列使用了函数

同样的道理,索引列使用了函数,一样会导致相同的后果

SELECT `sname` FROM `stu` WHERE concat(`sname`,'abc') ='Jaskeyabc'; -- 不会使用索引,因为使用了函数运算,原理与上面相同
SELECT `sname` FROM `stu` WHERE `sname` =concat('Jaskey','abc'); -- 会使用索引

索引列使用了Like %XXX

SELECT * FROM `houdunwang` WHERE `uname` LIKE '前缀就走索引%' -- 走索引
SELECT * FROM `houdunwang` WHERE `uname` LIKE '后缀不走索引%' -- 不走索引

所以当需要搜索email列中.com结尾的字符串而email上希望走索引时候,可以考虑数据库存储一个反向的内容reverse_email

SELECT * FROM `table` WHERE `reverse_email` LIKE REVERSE('%.com'); -- 走索引

注:以上如果你使用REVERSE(email) = REVERSE('%.com'),一样得不到你想要的结果,因为你在索引列email列上使用了函数,MySQL不会使用该列索引

同样的,索引列上使用正则表达式也不会走索引。

字符串列与数字直接比较

这是一个坑,假设有一张表,里面的a列是一个字符char类型,且a上建立了索引,你用它与数字类型做比较判断的话:

 CREATE TABLE `t1` (`a` char(10));

 SELECT * FROM `t1` WHERE `a`='1' -- 走索引
 SELECT * FROM `t2` WHERE `a`=1 -- 字符串和数字比较,不走索引!

但是如果那个表那个列是一个数字类型,拿来和字符类型的做比较,则不会影响到使用索引

 CREATE TABLE `t2` (`b` int);

 SELECT * FROM `t2` WHERE `b`='1' -- 虽然b是数字类型,和'1'比较依然走索引

但是,无论如何,这种额外的隐式类型转换都是开销,而且由于有字符和数字比就不走索引的情况,故建议避免一切隐式类型转换

尽量避免 OR 操作

select * from dept where dname='jaskey' or loc='bj' or deptno=45 --如果条件中有or,即使其中有条件带索引也不会使用。换言之,就是要求使用的所有字段,都必须建立索引

所以除非每个列都建立了索引,否则不建议使用OR,在多列OR中,可以考虑用UNION 替换

select * from dept where dname='jaskey' union
select * from dept where loc='bj' union
select * from dept where deptno=45

Git中只merge部分commit

cherry-pick

在Git 1.7.2以上的版本引入了一个 cheery-pick的命令可以只merge 部分的commit而不用直接把整个分支merge过来

git cherry-pick <commit 号>

如:

git cherry-pick e43a6fd3e94888d76779ad79fb568ed180e5fcdf

这样就只会把这个e43a6fd3e94888d76779ad79fb568ed180e5fcdf commit的内容pull到当前的分支,不过你会得到一个新的commit。 这样就可以按需merge需要的commit,而不需要的就可以直接废弃咯。

多个commit:

可以用空格指定多个commit:

git cherry-pick A B C D E F

CSS实现垂直居中

CSS垂直居中真是一个令人头疼的事,最近遇到了一个较为简单且通用的方法,总结如下:

1.无需要设置自己高度,和父容器高度, 利用绝对定位只需要以下三行:

parentElement{
    position:relative;
}
childElement{
    position: absolute;
    top: 50%;
    transform: translateY(-50%);
}

2.若只有父容器下只有一个元素,且父元素设置了高度,则只需要使用相对定位即可

parentElement{
    height:xxx;
}
.childElement {
  position: relative;
  top: 50%;
  transform: translateY(-50%);
}

例子猛击:DEMO


Flex布局

如果你不需要兼容老式浏览器(例如IE9及以下),使用Flex布局可以非常轻松实现

浏览器支持如下: Flex的支持性

样式:

parentElement{
    display:flex;/*Flex布局*/
    display: -webkit-flex; /* Safari */
    align-items:center;/*交叉轴居中,这里由于flex-direction默认是row,即垂直居中*/
}

注意,设为Flex布局以后,子元素的float、clear和vertical-align属性将失效

例子猛击:Flex demo


Flex教程可参考这里

Octopress 博文中引入javascript文件/HTML文件

.markdown文件中是可以引入javascript甚至是html文件的,这样以后博文里面插入代码运行结果写demo就很方便了。

以下是具体方法,并带一个例子

引入Javascript

语法:

<script type="text/javascript" src="/path/to/file.js"></script>

例如: 在 source 中的404.markdown 中 加入:

<script type="text/javascript" src="http://www.qq.com/404/search_children.js" charset="utf-8></script>

即可在404页面中跳转到腾讯的公益页面

引入HTML文件

以下为引入后的样本实例:

.markdown文件中,在需要引入HTML文件的地方写上:

’{‘%include demo/include_HTML_demo.html +’%’}’ (注:去掉其中的单引号)

HTML代码即会导入,若内嵌JavaScript脚本,也会自动导入允许。

这里你看到的区域都是外部导入的HTML文件,尝试点击这里,会发现有脚本运行.

点我试试看

注:由于octopress已经引入了jQuery,故本页面无须额外引入jQuery.

Jackson 操作JSON

Maven 支持:

  <repositories>
    <repository>
        <id>codehaus</id>
        <url>http://repository.codehaus.org/org/codehaus</url>
    </repository>
  </repositories>

  <dependencies>
    <dependency>
        <groupId>org.codehaus.jackson</groupId>
        <artifactId>jackson-mapper-asl</artifactId>
        <version>1.8.5</version>
    </dependency>
  </dependencies>

需要进行JSON操作的转换,仅需要一个ObjectMapper对象 ObjectMapper mapper = new ObjectMapper();

转换为对象时候,使用read相关方法,转换为JSON字符串时,使用write相关方法。

Java object to JSON

ObjectMapper mapper = new ObjectMapper();
mapper.writeValue(new File("c:\\user.json"), user);//写到文件,User 有get set 方法的POJO

大多数情况下,我们只需要一个JSON字符串,可以使用StringWriter作为参数的重载的writeValue方法:

StringWriter sw =new StringWriter();
ObjectMapper mapper = new ObjectMapper();
mapper.writeValue(sw, user);//写到StringWriter
String JSON = sw.toString();
sw.close();

也可以直接使用writeValueAsString ,其内部使用的也是StringWriter的。

ObjectMapper mapper = new ObjectMapper();
String JSON  = mapper.writeValueAsString(user));//直接转为String 类型的JSON

JSON to Java object

ObjectMapper mapper = new ObjectMapper();
User user = mapper.readValue(new File("c:\\user.json"), User.class);//User 有get set 方法的POJO

集合操作

Map/List to JSON

和普通对象一样,使用(同样,可以使用重载方法)

objectMapper.writeValue(new File(jsonFilePath), mapObject);
objectMapper.writeValue(new File(jsonFilePath), listObject);

JSON to Map/List/数组

     Map<String, Object> mapObject = mapper.readValue(new File('c:\\user.json'),
                new TypeReference<Map<String, Object>>() {});//使用TypeReference, 注意末尾有{}



     List<String> listObject = mapper.readValue(new File('c:\\user.json'),
                new TypeReference<List<String>(){});//使用TypeReference, 注意末尾有{}


     String[] array = mapper.readValue(new File('c:\\user.json'),
                String[].class);//使用class对象