一篇文章让你彻底掌握 Scala

Scala 是大数据领域的热门语言，如：Akka、Kafka，所以，想要学习大数据顶级开源项目的源码，有必要具备一定的 Scala 基础。

基本语法

Scala 基本语法需要注意以下几点：

• 区分大小写 - Scala 是大小写敏感的。
• 类名 - 对于所有的类名的第一个字母要大写。示例：class MyFirstScalaClass
• 方法名称 - 所有的方法名称的第一个字母用小写。示例：def myMethodName()
• 程序文件名 - 程序文件的名称应该与对象名称完全匹配(新版本不需要了，但建议保留这种习惯)。示例: 假设”HelloWorld”是对象的名称。那么该文件应保存为’HelloWorld.scala”
• def main(args: Array[String]) - Scala 程序从 main() 方法开始处理，这是每一个 Scala 程序的强制程序入口部分。
• 一行中只有空格或者带有注释，Scala 会认为其是空行，会忽略它。标记可以被空格或者注释来分割。
• Scala 是面向行的语言，语句可以用分号（;）结束或换行符。

注释

Scala 类似 Java 支持单行和多行注释。

【示例】单行和多行注释

1
2
3
4
5
6
7
8
9

object HelloWorld {
  /*
   * 这是我的第一个 Scala 程序
   * 以下程序将输出'Hello World!'
   */
  def main(args: Array[String]) {
    println("Hello, world!") // 输出 Hello World
  }
}

变量

在 Scala 中，使用关键词 var 声明变量，使用关键词 val 声明常量。

【示例】声明变量

1
2

var myVar : String = "Foo"
var myVar : String = "Too"

【示例】声明常量

1

val myVal : String = "Foo"

变量类型声明

变量的类型在变量名之后等号之前声明。定义变量的类型的语法格式如下：

1
2
3
4

// 声明变量类型
var VariableName : DataType [=  Initial Value]
// 声明常量类型
val VariableName : DataType [=  Initial Value]

在 Scala 中声明变量和常量不一定要指明数据类型，在没有指明数据类型的情况下，其数据类型是通过变量或常量的初始值推断出来的。所以，如果在没有指明数据类型的情况下声明变量或常量必须要给出其初始值，否则将会报错。

1
2
3

var myVar = 10;
val myVal = "Hello, Scala!";
val xmax, ymax = 100

数据类型

Scala 与 Java 有着相同的数据类型：

上表中列出的数据类型都是对象，也就是说 scala 没有 java 中的原生类型。在 scala 是可以对数字等基础类型调用方法的。

数组

Scala 数组声明的语法格式：

1
2
3
4

// 方式一
var z:Array[String] = new Array[String](3)
// 方式二
var z = new Array[String](3)

逻辑控制语句

条件语句

【示例】条件语句

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

object IfDemo {
  def main(args: Array[String]) {
    var x = 30;

    if (x == 10) {
      println("X 的值为 10");
    } else if (x == 20) {
      println("X 的值为 20");
    } else if (x == 30) {
      println("X 的值为 30");
    } else {
      println("无法判断 X 的值");
    }
  }
}

循环语句

和 Java 一样，Scala 支持 while、do ... while、for 三种循环语句。

1
2
3
4
5
6
7
8
9
10
11
12

object WhileDemo {
  def main(args: Array[String]) {
    // 局部变量
    var a = 10;

    // while 循环执行
    while (a < 20) {
      println("Value of a: " + a);
      a = a + 1;
    }
  }
}

**scala 不支持 break 和 continue**。但是，可以通过 Breaks 对象来进行循环控制。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

import scala.util.control._

object BreakDemo {
  def main(args: Array[String]) {
    var a = 0;
    var b = 0;
    val numList1 = List(1, 2, 3, 4, 5);
    val numList2 = List(11, 12, 13);

    val outer = new Breaks;
    val inner = new Breaks;

    outer.breakable {
      for (a <- numList1) {
        println("Value of a: " + a);
        inner.breakable {
          for (b <- numList2) {
            println("Value of b: " + b);
            if (b == 12) {
              inner.break;
            }
          }
        } // 内嵌循环中断
      }
    } // 外部循环中断
  }
}

模式匹配

scala 的 match 对应 Java 里的 switch，但是写在选择器表达式之后。即： 选择器 match {备选项}。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

/**
  * @author peng.zhang
  */
object MatchDemo {
  def main(args: Array[String]) {
    println(matchTest("two"))
    println(matchTest("test"))
    println(matchTest(1))
    println(matchTest(6))

  }

  def matchTest(x: Any): Any = x match {
    case 1      => "one"
    case "two"  => 2
    case y: Int => "scala.Int"
    case _      => "many"
  }
}

运算符

Scala 含有丰富的内置运算符，包括以下几种类型：

• 算术运算符：+、-、*、/、%
• 关系运算符：==、!=、>、<、>=、<=
• 逻辑运算符：&&、||、!
• 位运算符：~、&、|、^、<<、>>、>>> （无符号右移）
• 赋值运算符：=

方法与函数

Scala 有方法与函数，二者在语义上的区别很小。

Scala 中的方法跟 Java 的类似，方法是组成类的一部分。

Scala 中的函数则是一个完整的对象，Scala 中的函数其实就是继承了 Trait 的类的对象。

Scala 中使用 val 语句可以定义函数，def 语句定义方法。

【示例】

1
2
3
4

class Test {
  def m(x: Int) = x + 3
  val f = (x: Int) => x + 3
}

闭包

闭包是一个函数，返回值依赖于声明在函数外部的一个或多个变量。

闭包通常来讲可以简单的认为是可以访问一个函数里面局部变量的另外一个函数。

如下面这段匿名的函数：

1

val multiplier = (i:Int) => i * 10

函数体内有一个变量 i，它作为函数的一个参数。如下面的另一段代码：

1

val multiplier = (i:Int) => i * factor

在 multiplier 中有两个变量：i 和 factor。其中的一个 i 是函数的形式参数，在 multiplier 函数被调用时，i 被赋予一个新的值。然而，factor 不是形式参数，而是自由变量，考虑下面代码：

1
2

var factor = 3
val multiplier = (i:Int) => i * factor

这里我们引入一个自由变量 factor，这个变量定义在函数外面。

这样定义的函数变量 multiplier 成为一个”闭包”，因为它引用到函数外面定义的变量，定义这个函数的过程是将这个自由变量捕获而构成一个封闭的函数。

1
2
3
4
5
6
7
8
9

object ClosureDemo {
  def main(args: Array[String]) {
    println("muliplier(1) value = " + multiplier(1))
    println("muliplier(2) value = " + multiplier(2))
  }

  var factor = 3
  val multiplier = (i: Int) => i * factor
}

集合

Scala 集合支持 List、Set、Map、元祖、Option。

1
2
3
4
5
6
7
8
9
10
11
12
13
14

// 定义整型 List
val x = List(1,2,3,4)

// 定义 Set
val x = Set(1,3,5,7)

// 定义 Map
val x = Map("one" -> 1, "two" -> 2, "three" -> 3)

// 创建两个不同类型元素的元组
val x = (10, "Runoob")

// 定义 Option
val x:Option[Int] = Some(5)

迭代器

迭代器 it 的两个基本操作是 next 和 hasNext。

调用 it.next() 会返回迭代器的下一个元素，并且更新迭代器的状态。

调用 it.hasNext() 用于检测集合中是否还有元素。

1
2
3
4
5
6
7
8
9

object Test {
   def main(args: Array[String]) {
      val it = Iterator("Baidu", "Google", "Runoob", "Taobao")

      while (it.hasNext){
         println(it.next())
      }
   }
}

类和对象

类是对象的抽象，而对象是类的具体实例。类是抽象的，不占用内存，而对象是具体的，占用存储空间。类是用于创建对象的蓝图，它是一个定义包括在特定类型的对象中的方法和变量的软件模板。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

class Point(val xc: Int, val yc: Int) {
  var x: Int = xc
  var y: Int = yc

  def move(dx: Int, dy: Int) {
    x = x + dx
    y = y + dy
    println("x 的坐标点 : " + x);
    println("y 的坐标点 : " + y);
  }
}

class Location(override val xc: Int, override val yc: Int, val zc: Int)
    extends Point(xc, yc) {
  var z: Int = zc

  def move(dx: Int, dy: Int, dz: Int) {
    x = x + dx
    y = y + dy
    z = z + dz
    println("x 的坐标点 : " + x);
    println("y 的坐标点 : " + y);
    println("z 的坐标点 : " + z);
  }
}

object Test {
  def main(args: Array[String]) {
    val loc = new Location(10, 20, 15);

    // 移到一个新的位置
    loc.move(10, 10, 5);
  }
}

Trait

Scala Trait(特征) 相当于 Java 的接口，实际上它比接口还功能强大。

与接口不同的是，它还可以定义属性和方法的实现。

一般情况下 Scala 的类只能够继承单一父类，但是如果是 Trait(特征) 的话就可以继承多个，从结果来看就是实现了多重继承。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

trait Equal {
  def isEqual(x: Any): Boolean

  def isNotEqual(x: Any): Boolean = !isEqual(x)
}

class Point(xc: Int, yc: Int) extends Equal {
  var x: Int = xc
  var y: Int = yc

  def isEqual(obj: Any) =
    obj.isInstanceOf[Point] &&
      obj.asInstanceOf[Point].x == x
}

object Test {
  def main(args: Array[String]) {
    val p1 = new Point(2, 3)
    val p2 = new Point(2, 4)
    val p3 = new Point(3, 3)

    println(p1.isNotEqual(p2))
    println(p1.isNotEqual(p3))
    println(p1.isNotEqual(2))
  }
}

异常

Scala 抛出异常的方法和 Java 一样，使用 throw 关键词。

【示例】抛出异常

1

throw new IllegalArgumentException

【示例】捕获异常

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

import java.io.{FileNotFoundException, FileReader, IOException}

object Test {
  def main(args: Array[String]) {
    try {
      val f = new FileReader("input.txt")
    } catch {
      case ex: FileNotFoundException => {
        println("Missing file exception")
      }
      case ex: IOException => {
        println("IO Exception")
      }
    } finally {
      println("Exiting finally...")
    }
  }
}

输入输出

读取用户输入

使用 scala.io.StdIn.readLine() 方法读取用户输入

1
2
3
4
5
6
7
8

object StdInDemo {
  def main(args: Array[String]) {
    print("请输入内容: ")
    val line = StdIn.readLine()

    println("你输入的是: " + line)
  }
}

读取文件内容

1
2
3
4
5
6
7
8
9

object SourceDemo {
  def main(args: Array[String]) {
    println("文件内容为:")

    Source.fromFile("test.txt").foreach {
      print
    }
  }
}

包

定义包

Scala 使用 package 关键字定义包，在 Scala 将代码定义到某个包中有两种方式：

第一种方法和 Java 一样，在文件的头定义包名，这种方法就后续所有代码都放在该包中。 比如：

1
2

package com.runoob
class HelloWorld

第二种方法有些类似 C#，如：

1
2
3

package com.runoob {
  class HelloWorld
}

引用

Scala 使用 import 关键字引用包。

1
2
3
4
5
6
7

import java.awt.Color  // 引入Color

import java.awt._  // 引入包内所有成员

def handler(evt: event.ActionEvent) { // java.awt.event.ActionEvent
  ...  // 因为引入了java.awt，所以可以省去前面的部分
}

import 语句可以出现在任何地方，而不是只能在文件顶部。import 的效果从开始延伸到语句块的结束。这可以大幅减少名称冲突的可能性。

如果想要引入包中的几个成员，可以使用 selector（选取器）：

1
2
3
4
5
6
7

import java.awt.{Color, Font}

// 重命名成员
import java.util.{HashMap => JavaHashMap}

// 隐藏成员
import java.util.{HashMap => _, _} // 引入了util包的所有成员，但是HashMap被隐藏了

注意：默认情况下，Scala 总会引入 java.lang._ 、 scala._ 和 Predef._，这里也能解释，为什么以 scala 开头的包，在使用时都是省去 scala.的。

访问修饰符

Scala 访问修饰符基本和 Java 的一样，分别有：private，protected，public。

如果没有指定访问修饰符，默认情况下，Scala 对象的访问级别都是 public。

Scala 中的 private 限定符，比 Java 更严格，在嵌套类情况下，外层类甚至不能访问被嵌套类的私有成员。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

class Outer {

  class Inner {
    private def f() {
      println("f")
    }

    class InnerMost {
      f() // 正确
    }

  }

  (new Inner).f() //错误
}

参考资料

• Scala 官网
• Scala 菜鸟教程

Kafka 可靠传输

消息不丢失

如何保证消息的可靠性传输，或者说，如何保证消息不丢失？这对于任何 MQ 都是核心问题。

一条消息从生产到消费，可以划分三个阶段：

• 生产阶段：Producer 创建消息，并通过网络发送给 Broker。
• 存储阶段：Broker 收到消息并存储，如果是集群，还要同步副本给其他 Broker。
• 消费阶段：Consumer 向 Broker 请求消息，Broker 通过网络传输给 Consumer。

这三个阶段都可能丢失数据，所以要保证消息丢失，就需要任意一环都保证可靠。

存储阶段

存储阶段指的是 Kafka Server，也就是 Broker 如何保证消息不丢失。

一句话概括，Kafka 只对“已提交”的消息（committed message）做有限度的持久化保证。

上面的话可以解读为：

• 已提交：只有当消息被写入分区的若干同步副本时，才被认为是已提交的。为什么是若干个 Broker 呢？这取决于你对“已提交”的定义。你可以选择只要 Leader 成功保存该消息就算是已提交，也可以是令所有 Broker 都成功保存该消息才算是已提交。
• 持久化：Kafka 的数据存储在磁盘上，所以只要写入成功，天然就是持久化的。
• 只要还有一个副本是存活的，那么已提交的消息就不会丢失。
• 消费者只能读取已提交的消息。

副本机制

Kafka 的副本机制是 kafka 可靠性保证的核心。

Kafka 的主题被分为多个分区，分区是基本的数据块。每个分区可以有多个副本，有一个是 Leader（主副本），其他是 Follower（从副本）。所有数据都直接发送给 Leader，或者直接从 Leader 读取事件。Follower 只需要与 Leader 保持同步，并及时复制最新的数据。当 Leader 宕机时，从 Follower 中选举一个成为新的 Leader。

Broker 有 3 个配置参数会影响 Kafka 消息存储的可靠性。

副本数

replication.factor 的作用是设置每个分区的副本数。replication.factor 是主题级别配置； default.replication.factor 是 broker 级别配置。

副本数越多，数据可靠性越高；但由于副本数增多，也会增加同步副本的开销，可能会降低集群的可用性。一般，建议设为 3，这也是 Kafka 的默认值。

不完全的选主

unclean.leader.election.enable 用于控制是否支持不同步的副本参与选举 Leader。unclean.leader.election.enable 是 broker 级别（实际上是集群范围内）配置，默认值为 true。

• 如果设为 true，代表着允许不同步的副本成为主副本（即不完全的选举），那么将面临丢失消息的风险；
• 如果设为 false，就要等待原先的主副本重新上线，从而降低了可用性。

最少同步副本

min.insync.replicas 控制的是消息至少要被写入到多少个副本才算是“已提交”。min.insync.replicas 是主题级别和 broker 级别配置。

尽管可以为一个主题配置 3 个副本，但还是可能会出现只有一个同步副本的情况。如果这个同步副本变为不可用，则必须在可用性和数据一致性之间做出选择。Kafka 中，消息只有被写入到所有的同步副本之后才被认为是已提交的。但如果只有一个同步副本，那么在这个副本不可用时，则数据就会丢失。

如果要确保已经提交的数据被已写入不止一个副本，就需要把最小同步副本的设置为大一点的值。

注意：要确保 replication.factor > min.insync.replicas。如果两者相等，那么只要有一个副本挂机，整个分区就无法正常工作了。我们不仅要改善消息的持久性，防止数据丢失，还要在不降低可用性的基础上完成。推荐设置成 replication.factor = min.insync.replicas + 1。

生产阶段

在生产消息阶段，消息队列一般通过请求确认机制，来保证消息的可靠传递，Kafka 也不例外。

Kafka 生产 中提到了，Kafka 有三种发送方式：同步、异步、异步回调。

同步方式能保证消息不丢失，但性能太差；异步方式发送消息，通常会立即返回，但消息可能丢失。

解决生产者丢失消息的方案：

生产者使用异步回调方式 producer.send(msg, callback) 发送消息。callback（回调）能准确地告诉你消息是否真的提交成功了。一旦出现消息提交失败的情况，你就可以有针对性地进行处理。

• 如果是因为那些瞬时错误，那么仅仅让 Producer 重试就可以了；
• 如果是消息不合格造成的，那么可以调整消息格式后再次发送。

然后，需要基于以下几点来保证 Kafka 生产者的可靠性：

ACK

生产者可选的确认模式有三种：acks=0、acks=1、acks=all。

• acks=0、acks=1 都有丢失数据的风险。

• acks=all 意味着会等待所有同步副本都收到消息。再结合 min.insync.replicas ，就可以决定在得到确认响应前，至少有多少副本能够收到消息。

这是最保险的做法，但也会降低吞吐量。

重试

如果 broker 返回的错误可以通过重试来解决，生产者会自动处理这些错误。

• 可重试错误，如：LEADER_NOT_AVAILABLE，主副本不可用，可能过一段时间，集群就会选举出新的主副本，重试可以解决问题。
• 不可重试错误，如：INVALID_CONFIG，即使重试，也无法改变配置选项，重试没有意义。

需要注意的是：有时可能因为网络问题导致没有收到确认，但实际上消息已经写入成功。生产者会认为出现临时故障，重试发送消息，这样就会出现重复记录。所以，尽可能在业务上保证幂等性。

设置 retries 为一个较大的值。这里的 retries 同样是 Producer 的参数，对应前面提到的 Producer 自动重试。当出现网络的瞬时抖动时，消息发送可能会失败，此时配置了 retries > 0 的 Producer 能够自动重试消息发送，避免消息丢失。

错误处理

开发者需要自行处理的错误：

• 不可重试的 broker 错误，如消息大小错误、认证错误等；
• 消息发送前发生的错误，如序列化错误；
• 生产者达到重试次数上限或消息占用的内存达到上限时发生的错误。

消费阶段

前文已经提到，消费者只能读取已提交的消息。这就保证了消费者接收到消息时已经具备了数据一致性。

消费者唯一要做的是确保哪些消息是已经读取过的，哪些是没有读取过的（通过提交偏移量给 Broker 来确认）。如果消费者提交了偏移量却未能处理完消息，那么就有可能造成消息丢失，这也是消费者丢失消息的主要原因。

消费者的可靠性配置

• group.id - 如果希望消费者可以看到主题的所有消息，那么需要为它们设置唯一的 group.id。
• auto.offset.reset - 有两个选项：
  • earliest - 消费者会从分区的开始位置读取数据
  • latest - 消费者会从分区末尾位置读取数据
• enable.auto.commit - 消费者自动提交偏移量。如果设为 true，处理流程更简单，但无法保证重复处理消息。
• auto.commit.interval.ms - 自动提交的频率，默认为每 5 秒提交一次。

显示提交偏移量

如果 enable.auto.commit 设为 true，即自动提交，就无需考虑提交偏移量的问题。

如果选择显示提交偏移量，需要考虑以下问题：

• 必须在处理完消息后再发送确认（提交偏移量），不要收到消息立即确认。
• 提交频率是性能和重复消息数之间的权衡
• 分区再均衡
• 消费可能需要重试机制
• 超时处理
• 消费者可能需要维护消费状态，如：处理完消息后，记录在数据库中。
• 幂等性设计
  • 写数据库：根据主键判断记录是否存在
  • 写 Redis：set 操作天然具有幂等性
  • 复杂的逻辑处理，则可以在消息中加入全局 ID

重复消息

在 MQTT 协议中，给出了三种传递消息时能够提供的服务质量标准，这三种服务质量从低到高依次是：

• At most once：至多一次。消息在传递时，最多会被送达一次。换一个说法就是，没什么消息可靠性保证，允许丢消息。一般都是一些对消息可靠性要求不太高的监控场景使用，比如每分钟上报一次机房温度数据，可以接受数据少量丢失。
• At least once: 至少一次。消息在传递时，至少会被送达一次。也就是说，不允许丢消息，但是允许有少量重复消息出现。
• Exactly once：恰好一次。消息在传递时，只会被送达一次，不允许丢失也不允许重复，这个是最高的等级。

绝大部分消息队列提供的服务质量都是 At least once，包括 RocketMQ、RabbitMQ 和 Kafka 都是这样。也就是说，消息队列很难保证消息不重复。

一般解决重复消息的办法是，在消费端，保证消费消息的操作具备幂等性。

常用的实现幂等操作的方法：

利用数据库的唯一约束实现幂等

关系型数据库可以使用 INSERT IF NOT EXIST 语句防止重复；Redis 可以使用 SETNX 命令来防止重复；其他数据库只要支持类似语义，也是一个道理。

为更新的数据设置前置条件

如果满足条件就更新数据，否则拒绝更新数据，在更新数据的时候，同时变更前置条件中需要判断的数据。这样，重复执行这个操作时，由于第一次更新数据的时候已经变更了前置条件中需要判断的数据，不满足前置条件，则不会重复执行更新数据操作。

但是，如果我们要更新的数据不是数值，或者我们要做一个比较复杂的更新操作怎么办？用什么作为前置判断条件呢？更加通用的方法是，给数据增加一个版本号属性，每次更数据前，比较当前数据的版本号是否和消息中的版本号一致，如果不一致就拒绝更新数据，更新数据的同时将版本号 +1，一样可以实现幂等更新。

记录并检查操作

还有一种通用性最强，适用范围最广的实现幂等性方法：记录并检查操作，也称为“Token 机制或者 GUID（全局唯一 ID）机制”，实现的思路特别简单：在执行数据更新操作之前，先检查一下是否执行过这个更新操作。

具体的实现方法是，在发送消息时，给每条消息指定一个全局唯一的 ID，消费时，先根据这个 ID 检查这条消息是否有被消费过，如果没有消费过，才更新数据，然后将消费状态置为已消费。

需要注意的是，“检查消费状态，然后更新数据并且设置消费状态”中，三个操作必须作为一组操作保证原子性，才能真正实现幂等，否则就会出现 Bug。这一组操作可以通过分布式事务或分布式锁来保证其原子性。

消息的有序性

某些场景下，可能会要求按序发送消息。

方案一、单 Partition

Kafka 每一个 Partition 只能隶属于消费者群组中的一个 Consumer，换句话说，每个 Partition 只能被一个 Consumer 消费。所以，如果 Topic 是单 Partition，自然是有序的。

方案分析

优点：简单粗暴。开发者什么也不用做。

缺点：Kafka 基于 Partition 实现其高并发能力，如果使用单 Partition，会严重限制 Kafka 的吞吐量。

结论：作为分布式消息引擎，限制并发能力，显然等同于自废武功，所以，这个方案几乎是不可接受的。

方案二、同一个 key 的消息发送给指定 Partition

（1）生产者端显示指定 key 发往一个指定的 Partition，就可以保证同一个 key 在这个 Partition 中是有序的。

（2）接下来，消费者端为每个 key 设定一个缓存队列，然后让一个独立线程负责消费指定 key 的队列，这就保证了消费消息也是有序的。

消息积压

先修复消费者，然后停掉当前所有消费者。

新建 Topic，扩大分区，以提高并发处理能力。

创建临时消费者程序，并部署在多节点上，扩大消费处理能力。

最后处理完积压消息后，恢复原先部署架构。

验证系统可靠性

建议从 3 个层面验证系统的可靠性：

• 配置验证
• 应用验证
  • 客户端和服务器断开连接
  • 选举
  • 依次重启 broker
  • 依次重启生产者
  • 依次重启消费者
• 监控可靠性
  • 对于生产者来说，最重要的两个指标是消息的 error-rate 和 retry-rate。如果这两个指标上升，说明系统出了问题。
  • 对于消费者来说，最重要的指标是 consumer-lag，该指标表明了消费者的处理速度与最近提交到分区里的偏移量之间还有多少差距。

最佳实践

生产者

1. 不要使用 producer.send(msg)，而要使用 producer.send(msg, callback)。记住，一定要使用带有回调通知的 send 方法。
2. 设置 acks = all。acks 是 Producer 的一个参数，代表了你对“已提交”消息的定义。如果设置成 all，则表明所有副本 Broker 都要接收到消息，该消息才算是“已提交”。这是最高等级的“已提交”定义。
3. 设置 retries 为一个较大的值。这里的 retries 同样是 Producer 的参数，对应前面提到的 Producer 自动重试。当出现网络的瞬时抖动时，消息发送可能会失败，此时配置了 retries > 0 的 Producer 能够自动重试消息发送，避免消息丢失。

服务器（Kafka Broker）

1. 设置 unclean.leader.election.enable = false。这是 Broker 端的参数，它控制的是哪些 Broker 有资格竞选分区的 Leader。如果一个 Broker 落后原先的 Leader 太多，那么它一旦成为新的 Leader，必然会造成消息的丢失。故一般都要将该参数设置成 false，即不允许这种情况的发生。
2. 设置 replication.factor >= 3。这也是 Broker 端的参数。其实这里想表述的是，最好将消息多保存几份，毕竟目前防止消息丢失的主要机制就是冗余。
3. 设置 min.insync.replicas > 1。这依然是 Broker 端参数，控制的是消息至少要被写入到多少个副本才算是“已提交”。设置成大于 1 可以提升消息持久性。在实际环境中千万不要使用默认值 1。
4. 确保 replication.factor > min.insync.replicas。如果两者相等，那么只要有一个副本挂机，整个分区就无法正常工作了。我们不仅要改善消息的持久性，防止数据丢失，还要在不降低可用性的基础上完成。推荐设置成 replication.factor = min.insync.replicas + 1。

消费者

1. 确保消息消费完成再提交。Consumer 端有个参数 enable.auto.commit，最好把它设置成 false，并采用手动提交位移的方式。就像前面说的，这对于单 Consumer 多线程处理的场景而言是至关重要的。

参考资料

• 官方
  • Kafka 官网
  • Kafka Github
  • Kafka 官方文档
• 书籍
  • 《Kafka 权威指南》
• 教程
  • 消息队列高手课
  • Kafka 中文文档
  • Kafka 核心技术与实战

Kafka 消费

消费者简介

获取消息模式

消息引擎获取消息有两种模式：

• push 模式 - MQ 推送数据给消费者
• pull 模式 - 消费者主动向 MQ 请求数据

Kafka 消费者（Consumer）以 pull 方式从 Broker 拉取消息。相比于 push 方式，pull 方式灵活度和扩展性更好，因为消费的主动性由消费者自身控制。

push 模式的优缺点：

• 缺点：由 broker 决定消息推送的速率，对于不同消费速率的 consumer 就不太好处理了。push 模式下，当 broker 推送的速率远大于 consumer 消费的速率时，consumer 恐怕就要崩溃了。

push 模式的优缺点：

• 优点：consumer 可以根据自己的消费能力自主的决定消费策略
• 缺点：如果 broker 没有可供消费的消息，将导致 consumer 不断在循环中轮询，直到新消息到达。为了避免这点，Kafka 有个参数可以让 consumer 阻塞直到新消息到达

消费者

每个 Consumer 的唯一元数据是该 Consumer 在日志中消费的位置。这个偏移量是由 Consumer 控制的：Consumer 通常会在读取记录时线性的增加其偏移量。但实际上，由于位置由 Consumer 控制，所以 Consumer 可以采用任何顺序来消费记录。

一条消息只有被提交，才会被消费者获取到。如下图，只能消费 Message0、Message1、Message2：

消费者群组

Consumer Group 是 Kafka 提供的可扩展且具有容错性的消费者机制。

Kafka 的写入数据量很庞大，如果只有一个消费者，消费消息速度很慢，时间长了，就会造成数据积压。为了减少数据积压，Kafka 支持消费者群组，可以让多个消费者并发消费消息，对数据进行分流。

Kafka 消费者从属于消费者群组，一个群组里的 Consumer 订阅同一个 Topic，一个主题有多个 Partition，每一个 Partition 只能隶属于消费者群组中的一个 Consumer。

如果超过主题的分区数量，那么有一部分消费者就会被闲置，不会接收到任何消息。

同一时刻，一条消息只能被同一消费者组中的一个消费者实例消费。

不同消费者群组之间互不影响。

消费流程

Kafka 消费者通过 poll 模式来获取消息，但是获取消息时并不是立刻返回结果，需要考虑两个因素：

• 消费者通过 customer.poll(time) 中设置等待时间
• Broker 会等待累计一定量数据，然后发送给消费者。这样可以减少网络开销。

poll 除了获取消息外，还有其他作用：

• 发送心跳信息。消费者通过向被指派为群组协调器的 Broker 发送心跳来维护他和群组的从属关系，当机器宕掉后，群组协调器触发再均衡。

消费者 API

创建消费者

1
2
3
4
5
6
7
8
9
10
11
12

Properties props = new Properties();
// 服务器地址
props.put("bootstrap.servers", "localhost:9092");
// 消费者群组
props.put("group.id", "test");
// 关闭自动提交偏移量
props.put("enable.auto.commit", "false");
// 设置 key 反序列化器
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
// 设置 value 反序列化器
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

订阅主题

1
2
3
4

// 订阅主题列表
consumer.subscribe(Arrays.asList("t1", "t2"));
// 订阅所有与 test 相关的主题
consumer.subscribe("test.*");

subscribe 方法允许传入一个正则表达式，这样就可以匹配多个主题。如果有人创建了新的主题，并且主题名恰好匹配正则表达式，那么会立即触发一次分区再均衡，消费者就可以读取新添加的主题。

轮询获取消息

消息轮询是消费者 API 的核心。一旦消费者订阅了主题，轮询就会处理所有细节，包括：群组协调、分区再均衡、发送心跳和获取数据。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

try {
    // 3. 轮询
    while (true) {
        // 4. 消费消息
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            log.debug("topic = {}, partition = {}, offset = {}, key = {}, value = {}",
                record.topic(), record.partition(),
                record.offset(), record.key(), record.value());
        }
    }
} finally {
    // 5. 退出程序前，关闭消费者
    consumer.close();
}

手动提交偏移量

（1）同步提交

使用 commitSync() 提交偏移量最简单也最可靠。这个 API 会提交由 poll() 方法返回的最新偏移量，提交成功后马上返回，如果提交失败就抛出异常。

1
2
3
4
5
6
7
8
9
10
11
12
13

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("topic = %s, partition = %s, offset = %d, customer = %s, country = %s\n",
            record.topic(), record.partition(),
            record.offset(), record.key(), record.value());
    }
    try {
        consumer.commitSync();
    } catch (CommitFailedException e) {
        log.error("commit failed", e)
    }
}

同步提交的缺点：同步提交方式会一直阻塞，直到接收到 Broker 的响应请求，这会大大限制吞吐量。

（2）异步提交

在成功提交或碰到无法恢复的错误之前，commitSync() 会一直重试，但是 commitAsync() 不会，这也是 commitAsync() 不好的一个地方。它之所以不进行重试，是因为在它收到服务器响应的时候，可能有一个更大的偏移量已经提交成功。假设我们发出一个请求用于提交偏移量 2000，这个时候发生了短暂的通信问题，服务器收不到请求，自然也不会作出任何响应。与此同时，我们处理了另外一批消息，并成功提交了偏移量 3000。如果 commitAsync() 重新尝试提交偏移量 2000，它有可能在偏移量 3000 之后提交成功。这个时候如果发生再均衡，就会出现重复消息。

1
2
3
4
5
6
7
8
9

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("topic = %s, partition = %s, offset = % d, customer = %s, country = %s\n ",
            record.topic(), record.partition(), record.offset(),
            record.key(), record.value());
    }
    consumer.commitAsync();
}

commitAsync() 也支持回调，在 Broker 作出响应时会执行回调。回调经常被用于记录提交错误或生成度量指标，不过如果要用它来进行重试，则一定要注意提交的顺序。

1
2
3
4
5
6
7
8
9
10
11
12
13
14

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("topic = %s, partition = %s, offset = % d, customer = %s, country = %s\n ",
            record.topic(), record.partition(), record.offset(),
            record.key(), record.value());
    }
    consumer.commitAsync(new OffsetCommitCallback() {
        @Override
        public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception e) {
            if (e != null) { log.error("Commit failed for offsets {}", offsets, e); }
        }
    });
}

重试异步提交

可以使用一个单调递增的序列号来维护异步提交的顺序。在每次提交偏移量之后或在回调里提交偏移量时递增序列号。在进行重试前，先检查回调的序列号和即将提交的偏移量是否相等，如果相等，说明没有新的提交，那么可以安全地进行重试；如果序列号比较大，说明有一个新的提交已经发送出去了，应该停止重试。

（3）同步和异步组合提交

一般情况下，针对偶尔出现的提交失败，不进行重试不会有太大问题，因为如果提交失败是因为临时问题导致的，那么后续的提交总会有成功的。但如果这是发生在关闭消费者或再均衡前的最后一次提交，就要确保能够提交成功。

因此，在消费者关闭前一般会组合使用 commitSync() 和 commitAsync()。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("topic = %s, partition = %s, offset = % d, customer = %s, country = %s\n ",
                record.topic(), record.partition(), record.offset(), record.key(), record.value());
        }
        consumer.commitAsync();
    }
} catch (Exception e) {
    log.error("Unexpected error", e);
} finally {
    try {
        consumer.commitSync();
    } finally {
        consumer.close();
    }
}

（4）提交特定的偏移量

提交偏移量的频率和处理消息批次的频率是一样的。如果想要更频繁地提交该怎么办？如果 poll() 方法返回一大批数据，为了避免因再均衡引起的重复处理整批消息，想要在批次中间提交偏移量该怎么办？这种情况无法通过调用 commitSync() 或 commitAsync() 来实现，因为它们只会提交最后一个偏移量，而此时该批次里的消息还没有处理完。

解决办法是：消费者 API 允许在调用 commitSync() 和 commitAsync() 方法时传进去希望提交的分区和偏移量的 map。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

private int count = 0;
private final Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();


while (true) {
  ConsumerRecords<String, String> records = consumer.poll(100);
  for (ConsumerRecord<String, String> record : records) {
    System.out.printf("topic = %s, partition = %s, offset = % d, customer = %s, country = %s\n ",
                      record.topic(), record.partition(), record.offset(), record.key(), record.value());

    currentOffsets.put(new TopicPartition(record.topic(),
                                          record.partition()), new
                       OffsetAndMetadata(record.offset() + 1, "no metadata"));
    if (count % 1000 == 0) { consumer.commitAsync(currentOffsets, null); }
    count++;
  }
}

（5）从特定偏移量处开始处理

使用 poll() 方法可以从各个分区的最新偏移量处开始处理消息。

不过，有时候，我们可能需要从特定偏移量处开始处理消息。

• 从分区的起始位置开始读消息：seekToBeginning(Collection<TopicPartition> partitions) 方法
• 从分区的末尾位置开始读消息：seekToEnd(Collection<TopicPartition> partitions) 方法
• 查找偏移量：seek(TopicPartition partition, long offset) 方法

通过 seek(TopicPartition partition, long offset) 可以实现处理消息和提交偏移量在一个事务中完成。思路就是需要在客户端建立一张数据表，保证处理消息和和消息偏移量位置写入到这张数据表。在一个事务中，此时就可以保证处理消息和记录偏移量要么同时成功，要么同时失败。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

consumer.subscribe(topic);
// 1.第一次调用pool,加入消费者群组
consumer.poll(0);
// 2.获取负责的分区，并从本地数据库读取改分区最新偏移量，并通过seek方法修改poll获取消息的位置
for (TopicPartition partition: consumer.assignment())
    consumer.seek(partition, getOffsetFromDB(partition));

while (true) {
    ConsumerRecords<String, String> records =
    consumer.poll(100);
    for (ConsumerRecord<String, String> record : records)
    {
        processRecord(record);
        storeRecordInDB(record);
        storeOffsetInDB(record.topic(), record.partition(),
        record.offset());
    }
    commitDBTransaction();
}

关闭连接

如果想让消费者从轮询消费消息的无限循环中退出，可以通过另一个线程调用 consumer.wakeup() 方法。 consumer.wakeup() 是消费者唯一一个可以从其他线程里安全调用的方法。调用 consumer.wakeup() 可以退出 poll() ，并抛出 WakeupException 异常，或者如果调用 consumer.wakeup() 时线程没有等待轮询，那么异常将在下一轮调用 poll() 时抛出。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

Runtime.getRuntime().addShutdownHook(new Thread() {
    public void run() {
        System.out.println("Starting exit...");
        consumer.wakeup();
        try {
            mainThread.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
});

...

try {
    // looping until ctrl-c, the shutdown hook will cleanup on exit
    while (true) {
        ConsumerRecords<String, String> records =
            movingAvg.consumer.poll(1000);
        System.out.println(System.currentTimeMillis() +
            "--  waiting for data...");
        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("offset = %d, key = %s, value = %s\n",
                record.offset(), record.key(), record.value());
        }
        for (TopicPartition tp: consumer.assignment())
            System.out.println("Committing offset at position:" +
                consumer.position(tp));
            movingAvg.consumer.commitSync();
    }
} catch (WakeupException e) {
    // ignore for shutdown
} finally {
    consumer.close();
    System.out.println("Closed consumer and we are done");
}

分区再均衡

什么是分区再均衡

分区的所有权从一个消费者转移到另一个消费者，这样的行为被称为分区再均衡（Rebalance）。Rebalance 实现了消费者群组的高可用性和伸缩性。

Rebalance 本质上是一种协议，规定了一个 Consumer Group 下的所有 Consumer 如何达成一致，来分配订阅 Topic 的每个分区。比如某个 Group 下有 20 个 Consumer 实例，它订阅了一个具有 100 个分区的 Topic。正常情况下，Kafka 平均会为每个 Consumer 分配 5 个分区。这个分配的过程就叫 Rebalance。

当在群组里面新增/移除消费者或者新增/移除 kafka 集群 broker 节点时，群组协调器 Broker 会触发再均衡，重新为每一个 Partition 分配消费者。Rebalance 期间，消费者无法读取消息，造成整个消费者群组一小段时间的不可用。

何时生分区再均衡

分区再均衡的触发时机有三种：

• 消费者群组成员数发生变更。比如有新的 Consumer 加入群组或者离开群组，或者是有 Consumer 实例崩溃被“踢出”群组。
  • 新增消费者。consumer 订阅主题之后，第一次执行 poll 方法
  • 移除消费者。执行 consumer.close() 操作或者消费客户端宕机，就不再通过 poll 向群组协调器发送心跳了，当群组协调器检测次消费者没有心跳，就会触发再均衡。
• 订阅主题数发生变更。Consumer Group 可以使用正则表达式的方式订阅主题，比如 consumer.subscribe(Pattern.compile(“t.*c”)) 就表明该 Group 订阅所有以字母 t 开头、字母 c 结尾的主题。在 Consumer Group 的运行过程中，你新创建了一个满足这样条件的主题，那么该 Group 就会发生 Rebalance。
• 订阅主题的分区数发生变更。Kafka 当前只能允许增加一个主题的分区数。当分区数增加时，就会触发订阅该主题的所有 Group 开启 Rebalance。
  • 新增 broker。如重启 broker 节点
  • 移除 broker。如 kill 掉 broker 节点。

分区再均衡的过程

Rebalance 是通过消费者群组中的称为“群主”消费者客户端进行的。

（1）选择群主

当消费者要加入群组时，会向群组协调器发送一个 JoinGroup 请求。第一个加入群组的消费者将成为“群主”。群主从协调器那里获取群组的活跃成员列表，并负责给每一个消费者分配分区。

所谓协调者，在 Kafka 中对应的术语是 Coordinator，它专门为 Consumer Group 服务，负责为 Group 执行 Rebalance 以及提供位移管理和组成员管理等。具体来讲，Consumer 端应用程序在提交位移时，其实是向 Coordinator 所在的 Broker 提交位移。同样地，当 Consumer 应用启动时，也是向 Coordinator 所在的 Broker 发送各种请求，然后由 Coordinator 负责执行消费者组的注册、成员管理记录等元数据管理操作。

（2）消费者通过向被指派为群组协调器（Coordinator）的 Broker 定期发送心跳来维持它们和群组的从属关系以及它们对分区的所有权。

（3）群主从群组协调器获取群组成员列表，然后给每一个消费者进行分配分区 Partition。有两种分配策略：Range 和 RoundRobin。

• Range 策略，就是把若干个连续的分区分配给消费者，如存在分区 1-5，假设有 3 个消费者，则消费者 1 负责分区 1-2,消费者 2 负责分区 3-4，消费者 3 负责分区 5。
• RoundRoin 策略，就是把所有分区逐个分给消费者，如存在分区 1-5，假设有 3 个消费者，则分区 1->消费 1，分区 2->消费者 2，分区 3>消费者 3，分区 4>消费者 1，分区 5->消费者 2。

（4）群主分配完成之后，把分配情况发送给群组协调器。

（5）群组协调器再把这些信息发送给消费者。每个消费者只能看到自己的分配信息，只有群主知道所有消费者的分配信息。

如何判定消费者已经死亡

消费者通过向被指定为群组协调器的 Broker 发送心跳来维持它们和群组的从属关系以及它们对分区的所有权关系。只要消费者以正常的时间间隔发送心跳，就被认为是活跃的。消费者会在轮询消息或提交偏移量时发送心跳。如果消费者超时未发送心跳，会话就会过期，群组协调器认定它已经死亡，就会触发一次再均衡。

当一个消费者要离开群组时，会通知协调器，协调器会立即触发一次再均衡，尽量降低处理停顿。

查找协调者

所有 Broker 在启动时，都会创建和开启相应的 Coordinator 组件。也就是说，所有 Broker 都有各自的 Coordinator 组件。那么，Consumer Group 如何确定为它服务的 Coordinator 在哪台 Broker 上呢？答案就在我们之前说过的 Kafka 内部位移主题 __consumer_offsets 身上。

目前，Kafka 为某个 Consumer Group 确定 Coordinator 所在的 Broker 的算法有 2 个步骤。

1. 第 1 步：确定由位移主题的哪个分区来保存该 Group 数据：partitionId=Math.abs(groupId.hashCode() % offsetsTopicPartitionCount)。

2. 第 2 步：找出该分区 Leader 副本所在的 Broker，该 Broker 即为对应的 Coordinator。

分区再均衡的问题

• 首先，Rebalance 过程对 Consumer Group 消费过程有极大的影响。在 Rebalance 过程中，所有 Consumer 实例都会停止消费，等待 Rebalance 完成。
• 其次，目前 Rebalance 的设计是所有 Consumer 实例共同参与，全部重新分配所有分区。其实更高效的做法是尽量减少分配方案的变动。
• 最后，Rebalance 实在是太慢了。

避免分区再均衡

通过前文，我们已经知道了：分区再均衡的代价很高，应该尽量避免不必要的分区再均衡，以整体提高 Consumer 的吞吐量。

分区再均衡发生的时机有三个：

• 组成员数量发生变化
• 订阅主题数量发生变化
• 订阅主题的分区数发生变化

后面两个通常都是运维的主动操作，所以它们引发的 Rebalance 大都是不可避免的。实际上，大部分情况下，导致分区再均衡的原因是：组成员数量发生变化。

有两种情况，消费者并没有宕机，但也被视为消亡：

• 未及时发送心跳
• Consumer 消费时间过长

未及时发送心跳

第一类非必要 Rebalance 是因为未能及时发送心跳，导致 Consumer 被“踢出”Group 而引发的。因此，需要合理设置会话超时时间。这里给出一些推荐数值，你可以“无脑”地应用在你的生产环境中。

• 设置 session.timeout.ms = 6s。
• 设置 heartbeat.interval.ms = 2s。
• 要保证 Consumer 实例在被判定为“dead”之前，能够发送至少 3 轮的心跳请求，即 session.timeout.ms >= 3 * heartbeat.interval.ms。

将 session.timeout.ms 设置成 6s 主要是为了让 Coordinator 能够更快地定位已经挂掉的 Consumer。毕竟，我们还是希望能尽快揪出那些“尸位素餐”的 Consumer，早日把它们踢出 Group。希望这份配置能够较好地帮助你规避第一类“不必要”的 Rebalance。

Consumer 消费时间过长

第二类非必要 Rebalance 是 Consumer 消费时间过长导致的。此时，**max.poll.interval.ms** 参数值的设置显得尤为关键。如果要避免非预期的 Rebalance，你最好将该参数值设置得大一点，比你的下游最大处理时间稍长一点。

GC 参数

如果你按照上面的推荐数值恰当地设置了这几个参数，却发现还是出现了 Rebalance，那么我建议你去排查一下Consumer 端的 GC 表现，比如是否出现了频繁的 Full GC 导致的长时间停顿，从而引发了 Rebalance。为什么特意说 GC？那是因为在实际场景中，我见过太多因为 GC 设置不合理导致程序频发 Full GC 而引发的非预期 Rebalance 了。

提交偏移量

每次调用 poll() 方法，它总是会返回由生产者写入 Kafka 但还没有被消费者读取过的记录，Kafka 因此可以追踪哪些记录是被哪个群组的哪个消费者读取的。

更新分区当前位置的操作叫作提交。

偏移量的用处

如果消费者一直处于运行状态，那么偏移量就没有什么用处。不过，如果消费者发生崩溃或有新的消费者加入群组，就会触发再均衡，完成再均衡后，每个消费者可能分配到新的分区，而不是之前处理的那个。为了能够继续之前的工作，消费者需要读取每个分区最后一次提交的偏移量，然后从偏移量指定的地方继续处理。

（1）如果提交的偏移量小于客户端处理的最后一个消息的偏移量，那么处于两个偏移量之间的消息就会被重复处理。

（2）如果提交的偏移量大于客户端处理的最后一个消息的偏移量，那么处于两个偏移量之间的消息将会丢失。

由此可知，处理偏移量，会对客户端处理数据产生影响。

提交偏移量的旧方案

老版本的 Consumer Group 把偏移量保存在 ZooKeeper 中。ZooKeeper 是一个分布式的协调服务框架，Kafka 重度依赖它实现各种各样的协调管理。将偏移量保存在 ZooKeeper 外部系统的做法，最显而易见的好处就是减少了 Kafka Broker 端的状态保存开销，有利于实现伸缩性。

这种方案的问题在于：ZooKeeper 其实并不适合进行高频的写操作，而 Consumer Group 的偏移量更新却是一个非常频繁的操作。这种大吞吐量的写操作会极大地拖慢 ZooKeeper 集群的性能，因此 Kafka 社区渐渐有了这样的共识：将 Consumer 偏移量保存在 ZooKeeper 中是不合适的做法。

提交偏移量的新方案

新版本 Consumer 的偏移量管理机制其实也很简单。

消费者向一个叫做 _consumer_offsets 的特殊主题发送消息，消息里包含每个分区的偏移量。如果消费者一直处于运行状态，那么偏移量就没有什么用处。不过，如果消费者发生崩溃或有新的消费者加入群组，就会触发再均衡，完成再均衡后，每个消费者可能分配到新的分区，而不是之前处理的那个。为了能够继续之前的工作，消费者需要读取每个分区最后一次提交的偏移量，然后从偏移量指定的地方继续处理。

**_consumer_offsets 主题的 Key 中应该保存 3 部分内容：<Group ID，主题名，分区号 >**。

通常来说，当 Kafka 集群中的第一个 Consumer 程序启动时，Kafka 会自动创建偏移量主题。偏移量主题就是普通的 Kafka 主题，那么它自然也有对应的分区数。如果偏移量主题是 Kafka 自动创建的，那么该主题的分区数是 50，副本数是 3。分区数可以通过 offsets.topic.num.partitions 设置；副本数可以通过 offsets.topic.replication.factor 设置。

自动提交

自动提交是 Kafka 处理偏移量最简单的方式。

当 enable.auto.commit 属性被设为 true，那么每过 5s，消费者会自动把从 poll() 方法接收到的最大偏移量提交上去。提交时间间隔由 auto.commit.interval.ms 控制，默认值是 5s。

与消费者里的其他东西一样，自动提交也是在轮询里进行的。消费者每次在进行轮询时会检查是否该提交偏移量了，如果是，那么就会提交从上一次轮询返回的偏移量。

假设我们仍然使用默认的 5s 提交时间间隔，在最近一次提交之后的 3s 发生了再均衡，再均衡之后，消费者从最后一次提交的偏移量位置开始读取消息。这个时候偏移量已经落后了 3s（因为没有达到 5s 的时限，并没有提交偏移量），所以在这 3s 的数据将会被重复处理。虽然可以通过修改提交时间间隔来更频繁地提交偏移量，减小可能出现重复消息的时间窗的时间跨度，不过这种情况是无法完全避免的。

在使用自动提交时，每次调用轮询方法都会把上一次调用返回的偏移量提交上去，它并不知道具体哪些消息已经被处理了，所以在再次调用之前最好确保所有当前调用返回的消息都已经处理完毕（在调用 close() 方法之前也会进行自动提交）。一般情况下不会有什么问题，不过在处理异常或提前退出轮询时要格外小心。

自动提交虽然方便，不过无法避免丢失消息和分区再均衡时重复消息的问题。

手动提交

自动提交虽然方便，不过无法避免丢失消息和分区再均衡时重复消息的问题。因此，可以通过手动提交偏移量，由开发者自行控制。

首先，把 enable.auto.commit 设为 false，关闭自动提交。

如果 Kafka 触发了再均衡，我们需要在消费者失去对一个分区的所有权之前提交最后一个已处理记录的偏移量。如果消费者准备了一个缓冲区用于处理偶发的事件，那么在失去分区所有权之前，需要处理在缓冲区累积下来的记录。可能还需要关闭文件句柄、数据库连接等。

在为消费者分配新分区或移除旧分区时，可以通过消费者 API 执行一些应用程序代码，在调用 subscribe() 方法时传进去一个 ConsumerRebalanceListener 实例就可以了。 ConsumerRebalanceListener 有两个需要实现的方法。

• public void onPartitionsRevoked(Collection partitions) 方法会在再均衡开始之前和消费者停止读取消息之后被调用。如果在这里提交偏移量，下一个接管分区的消费者就知道该从哪里开始读取了。
• public void onPartitionsAssigned(Collection partitions) 方法会在重新分配分区之后和消费者开始读取消息之前被调用。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

private Map<TopicPartition, OffsetAndMetadata> currentOffsets=
  new HashMap<>();

private class HandleRebalance implements ConsumerRebalanceListener {
    public void onPartitionsAssigned(Collection<TopicPartition>
      partitions) {
    }

    public void onPartitionsRevoked(Collection<TopicPartition>
      partitions) {
        System.out.println("Lost partitions in rebalance.
          Committing current
        offsets:" + currentOffsets);
        consumer.commitSync(currentOffsets);
    }
}

try {
    consumer.subscribe(topics, new HandleRebalance());

    while (true) {
        ConsumerRecords<String, String> records =
          consumer.poll(100);
        for (ConsumerRecord<String, String> record : records)
        {
            System.out.println("topic = %s, partition = %s, offset = %d,
             customer = %s, country = %s\n",
             record.topic(), record.partition(), record.offset(),
             record.key(), record.value());
             currentOffsets.put(new TopicPartition(record.topic(),
             record.partition()), new
             OffsetAndMetadata(record.offset()+1, "no metadata"));
        }
        consumer.commitAsync(currentOffsets, null);
    }
} catch (WakeupException e) {
    // 忽略异常，正在关闭消费者
} catch (Exception e) {
    log.error("Unexpected error", e);
} finally {
    try {
        consumer.commitSync(currentOffsets);
    } finally {
        consumer.close();
        System.out.println("Closed consumer and we are done");
    }
}

反序列化器

生产者需要用序列化器将 Java 对象转换成字节数组再发送给 Kafka；同理，消费者需要用反序列化器将从 Kafka 接收到的字节数组转换成 Java 对象。

独立消费者

通常，会有多个 Kafka 消费者组成群组，关注一个主题。

但可能存在这样的场景：只需要一个消费者从一个主题的所有分区或某个特定的分区读取数据。这时，就不需要消费者群组和再均衡了，只需要把主题或分区分配给消费者，然后开始读取消息并提交偏移量。

如果是这样，就不需要订阅主题，取而代之的是为自己分配分区。一个消费者可以订阅主题（并加入消费者群组），或为自己分配分区，但不能同时做这两件事。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

List<PartitionInfo> partitionInfos = null;
partitionInfos = consumer.partitionsFor("topic");

if (partitionInfos != null) {
    for (PartitionInfo partition : partitionInfos)
        partitions.add(new TopicPartition(partition.topic(),
            partition.partition()));
    consumer.assign(partitions);

    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(1000);

        for (ConsumerRecord<String, String> record: records) {
            System.out.printf("topic = %s, partition = %s, offset = %d,
                customer = %s, country = %s\n",
                record.topic(), record.partition(), record.offset(),
                record.key(), record.value());
        }
        consumer.commitSync();
    }
}

消费者的配置

• bootstrap.servers - Broker 集群地址，格式：ip1:port,ip2:port…，不需要设定全部的集群地址，设置两个或者两个以上即可。
• group.id - 消费者隶属的消费者组名称，如果为空会报异常，一般而言，这个参数要有一定的业务意义。
• fetch.min.bytes - 消费者获取记录的最小字节数。Kafka 会等到有足够的数据时才返回消息给消费者，以降低负载。
• fetch.max.wait.ms - Kafka 需要等待足够的数据才返回给消费者，如果一直没有足够的数据，消费者就会迟迟收不到消息。所以需要指定 Broker 的等待延迟，一旦超时，直接返回数据给消费者。
• max.partition.fetch.bytes - 指定了服务器从每个分区返回给消费者的最大字节数。默认为 1 MB。
• session.timeout.ms - 指定了消费者的心跳超时时间。如果消费者没有在有效时间内发送心跳给群组协调器，协调器会视消费者已经消亡，从而触发分区再均衡。默认为 3 秒。
• auto.offset.reset - 指定了消费者在读取一个没有偏移量的分区或偏移量无效的情况下，该如何处理。
  • latest - 表示在偏移量无效时，消费者将从最新的记录开始读取分区记录。
  • earliest - 表示在偏移量无效时，消费者将从起始位置读取分区记录。
• enable.auto.commit - 指定了是否自动提交消息偏移量，默认开启。
• partition.assignment.strategy - 消费者的分区分配策略。
  • Range - 表示会将主题的若干个连续的分区分配给消费者。
  • RoundRobin - 表示会将主题的所有分区按照轮询方式分配给消费者。
• client.id - 客户端标识。
• max.poll.records - 用于控制单次能获取到的记录数量。
• receive.buffer.bytes - 用于设置 Socket 接收消息缓冲区（SO_RECBUF）的大小，默认值为 64KB。如果设置为-1，则使用操作系统的默认值。
• send.buffer.bytes - 用于设置 Socket 发送消息缓冲区（SO_SNDBUF）的大小，默认值为 128KB。与 receive.buffer.bytes 参数一样，如果设置为-1，则使用操作系统的默认值。

参考资料

• 官方
  • Kafka 官网
  • Kafka Github
  • Kafka 官方文档
• 书籍
  • 《Kafka 权威指南》
• 教程
  • Kafka 中文文档
  • Kafka 核心技术与实战

Kafka 生产

生产者简介

不管是把 Kafka 作为消息队列系统、还是数据存储平台，总是需要一个可以向 Kafka 写入数据的生产者和一个可以从 Kafka 读取数据的消费者，或者是一个兼具两种角色的应用程序。

使用 Kafka 的场景很多，诉求也各有不同，主要有：是否允许丢失消息？是否接受重复消息？是否有严格的延迟和吞吐量要求？

不同的场景对于 Kafka 生产者 API 的使用和配置会有直接的影响。

生产者传输实体

Kafka Producer 发送的数据对象叫做 ProducerRecord ，它有 4 个关键参数：

• Topic - 主题
• Partition - 分区（非必填）
• Key - 键（非必填）
• Value - 值

生产者发送流程

Kafka 生产者发送消息流程：

（1）序列化 - 发送前，生产者要先把键和值序列化成字节数组，这样它们才能够在网络中传输。

（2）分区 - 数据被传给分区器。如果在 ProducerRecord 中已经指定了分区，那么分区器什么也不会做；否则，分区器会根据 ProducerRecord 的键来选择一个分区。选定分区后，生产者就知道该把消息发送给哪个主题的哪个分区。

（3）批次传输 - 接着，这条记录会被添加到一个记录批次中。这个批次中的所有消息都会被发送到相同的主题和分区上。有一个独立的线程负责将这些记录批次发送到相应 Broker 上。

• 批次，就是一组消息，这些消息属于同一个主题和分区。
• 发送时，会把消息分成批次传输，如果每次只发送一个消息，会占用大量的网路开销。

（4）响应 - 服务器收到消息会返回一个响应。

• 如果成功，则返回一个 RecordMetaData 对象，它包含了主题、分区、偏移量；
• 如果失败，则返回一个错误。生产者在收到错误后，可以进行重试，重试次数可以在配置中指定。失败一定次数后，就返回错误消息。

生产者向 Broker 发送消息时是怎么确定向哪一个 Broker 发送消息？

• 生产者会向任意 broker 发送一个元数据请求（MetadataRequest），获取到每一个分区对应的 Leader 信息，并缓存到本地。
• 生产者在发送消息时，会指定 Partition 或者通过 key 得到到一个 Partition，然后根据 Partition 从缓存中获取相应的 Leader 信息。

生产者 API

Kafka 的 Java 生产者 API 主要的对象就是 KafkaProducer。通常我们开发一个生产者的步骤有 4 步。

1. 构造生产者对象所需的参数对象。
2. 利用第 1 步的参数对象，创建 KafkaProducer 对象实例。
3. 使用 KafkaProducer 的 send 方法发送消息。
4. 调用 KafkaProducer 的 close 方法关闭生产者并释放各种系统资源。

创建生产者

Kafka 生产者核心配置：

• bootstrap.servers - 指定了 Producer 启动时要连接的 Broker 地址。注：如果你指定了 1000 个 Broker 连接信息，那么，Producer 启动时就会首先创建与这 1000 个 Broker 的 TCP 连接。在实际使用过程中，并不建议把集群中所有的 Broker 信息都配置到 bootstrap.servers 中，通常你指定 3 ～ 4 台就足以了。因为 Producer 一旦连接到集群中的任一台 Broker，就能拿到整个集群的 Broker 信息，故没必要为 bootstrap.servers 指定所有的 Broker。
• key.serializer - 键的序列化器。
• value.serializer - 值的序列化器。

1
2
3
4
5
6
7
8
9
10

// 指定生产者的配置
final Properties properties = new Properties();
properties.put("bootstrap.servers", "localhost:9092");
// 设置 key 的序列化器
properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 设置 value 的序列化器
properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// 使用配置初始化 Kafka 生产者
Producer<String, String> producer = new KafkaProducer<>(properties);

异步发送

直接发送消息，不关心消息是否到达。

这种方式吞吐量最高，但有小概率会丢失消息。

【示例】异步发送

1
2
3
4
5
6
7

ProducerRecord<String, String> record =
            new ProducerRecord<>("CustomerCountry", "Precision Products", "France");
try {
    producer.send(record);
} catch (Exception e) {
    e.printStackTrace();
}

同步发送

返回一个 Future 对象，调用 get() 方法，会一直阻塞等待 Broker 返回结果。

这是一种可靠传输方式，但吞吐量最差。

【示例】同步发送

1
2
3
4
5
6
7

ProducerRecord<String, String> record =
            new ProducerRecord<>("CustomerCountry", "Precision Products", "France");
try {
    producer.send(record).get();
} catch (Exception e) {
    e.printStackTrace();
}

异步响应发送

代码如下，异步方式相对于“发送并忽略返回”的方式的不同在于：在异步返回时可以执行一些操作，如：抛出异常、记录错误日志。

这是一个折中的方案，即兼顾吞吐量，也保证消息不丢失。

【示例】异步响应发送

首先，定义一个 callback：

1
2
3
4
5
6
7
8

private class DemoProducerCallback implements Callback {
      @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
           if (e != null) {
               e.printStackTrace();
             }
        }
}

然后，使用这个 callback：

1
2
3

ProducerRecord<String, String> record =
            new ProducerRecord<>("CustomerCountry", "Biomedical Materials", "USA");
producer.send(record, new DemoProducerCallback());

关闭连接

调用 producer.close() 方法可以关闭 Kafka 生产者连接。

1
2
3
4
5
6
7
8
9

Producer<String, String> producer = new KafkaProducer<>(properties);
try {
   producer.send(new ProducerRecord<>(topic, msg));
} catch (Exception e) {
    e.printStackTrace();
} finally {
    // 关闭连接
    producer.close();
}

生产者的连接

Apache Kafka 的所有通信都是基于 TCP 的。无论是生产者、消费者，还是 Broker 之间的通信都是如此。

选用 TCP 连接是由于 TCP 本身提供的一些高级功能，如多路复用请求以及同时轮询多个连接的能力。

何时创建 TCP 连接

Kafka 生产者创建连接有三个时机：

（1）在创建 KafkaProducer 实例时，生产者应用会在后台创建并启动一个名为 Sender 的线程，该 Sender 线程开始运行时，首先会创建与 bootstrap.servers 中所有 Broker 的 TCP 连接。

（2）当 Producer 更新集群的元数据信息之后，如果发现与某些 Broker 当前没有连接，那么它就会创建一个 TCP 连接。

• 场景一：当 Producer 尝试给一个不存在的主题发送消息时，Broker 会告诉 Producer 说这个主题不存在。此时 Producer 会发送 METADATA 请求给 Kafka 集群，去尝试获取最新的元数据信息。
• 场景二：Producer 通过 metadata.max.age.ms 参数定期地去更新元数据信息。该参数的默认值是 300000，即 5 分钟，也就是说不管集群那边是否有变化，Producer 每 5 分钟都会强制刷新一次元数据以保证它是最及时的数据。

（3）当要发送消息时，Producer 发现尚不存在与目标 Broker 的连接，会创建一个 TCP 连接。

何时关闭 TCP 连接

Producer 端关闭 TCP 连接的方式有两种：一种是用户主动关闭；一种是 Kafka 自动关闭。

主动关闭是指调用 producer.close() 方法来关闭生产者连接；甚至包括用户调用 kill -9 主动“杀掉”Producer 应用。

如果设置 Producer 端 connections.max.idle.ms 参数大于 0（默认为 9 分钟），意味着，在 connections.max.idle.ms 指定时间内，如果没有任何请求“流过”某个 TCP 连接，那么 Kafka 会主动帮你把该 TCP 连接关闭。如果设置该参数为 -1，TCP 连接将成为永久长连接。

值得注意的是，在第二种方式中，TCP 连接是在 Broker 端被关闭的，但其实这个 TCP 连接的发起方是客户端，因此在 TCP 看来，这属于被动关闭的场景，即 passive close。被动关闭的后果就是会产生大量的 CLOSE_WAIT 连接，因此 Producer 端或 Client 端没有机会显式地观测到此连接已被中断。

序列化

Kafka 内置了常用 Java 基础类型的序列化器，如：StringSerializer、IntegerSerializer、DoubleSerializer 等。

但如果要传输较为复杂的对象，推荐使用序列化性能更高的工具，如：Avro、Thrift、Protobuf 等。

使用方式是通过实现 org.apache.kafka.common.serialization.Serializer 接口来引入自定义的序列化器。

分区

什么是分区

Kafka 的数据结构采用三级结构，即：主题（Topic）、分区（Partition）、消息（Record）。

在 Kafka 中，任意一个 Topic 维护了一组 Partition 日志，如下所示：

每个 Partition 都是一个单调递增的、不可变的日志记录，以不断追加的方式写入数据。Partition 中的每条记录会被分配一个单调递增的 id 号，称为偏移量（Offset），用于唯一标识 Partition 内的每条记录。

为什么要分区

为什么 Kafka 的数据结构采用三级结构？

分区的作用就是提供负载均衡的能力，以实现系统的高伸缩性（Scalability）。

不同的分区能够被放置到不同节点的机器上，而数据的读写操作也都是针对分区这个粒度而进行的，这样每个节点的机器都能独立地执行各自分区的读写请求处理。并且，我们还可以通过添加新的机器节点来增加整体系统的吞吐量。

分区策略

所谓分区策略是决定生产者将消息发送到哪个分区的算法，也就是负载均衡算法。

前文中已经提到，Kafka 生产者发送消息使用的对象 ProducerRecord ，可以选填 Partition 和 Key。不过，大多数应用会用到 key。key 有两个作用：作为消息的附加信息；也可以用来决定消息该被写到 Topic 的哪个 Partition，拥有相同 key 的消息将被写入同一个 Partition。

如果 ProducerRecord 指定了 Partition，则分区器什么也不做，否则分区器会根据 key 选择一个 Partition 。

• 没有 key 时的分发逻辑：每隔 topic.metadata.refresh.interval.ms 的时间，随机选择一个 partition。这个时间窗口内的所有记录发送到这个 partition。发送数据出错后会重新选择一个 partition。
• 根据 key 分发：Kafka 的选择分区策略是：根据 key 求 hash 值，然后将 hash 值对 partition 数量求模。这里的关键点在于，同一个 key 总是被映射到同一个 Partition 上。所以，在选择分区时，Kafka 会使用 Topic 的所有 Partition ，而不仅仅是可用的 Partition。这意味着，如果写入数据的 Partition 是不可用的，那么就会出错。

自定义分区策略

如果 Kafka 的默认分区策略无法满足实际需要，可以自定义分区策略。需要显式地配置生产者端的参数 partitioner.class。这个参数该怎么设定呢？

首先，要实现 org.apache.kafka.clients.producer.Partitioner 接口。这个接口定义了两个方法：partition 和 close，通常只需要实现最重要的 partition 方法。我们来看看这个方法的方法签名：

1

int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);

这里的 topic、key、keyBytes、value和 valueBytes 都属于消息数据，cluster 则是集群信息（比如当前 Kafka 集群共有多少主题、多少 Broker 等）。Kafka 给你这么多信息，就是希望让你能够充分地利用这些信息对消息进行分区，计算出它要被发送到哪个分区中。

接着，设置 partitioner.class 参数为自定义类的全限定名，那么生产者程序就会按照你的代码逻辑对消息进行分区。

负载均衡算法常见的有：

• 随机算法
• 轮询算法
• 最小活跃数算法
• 源地址哈希算法

可以根据实际需要去实现。

压缩

Kafka 的消息格式

目前，Kafka 共有两大类消息格式，社区分别称之为 V1 版本和 V2 版本。V2 版本是 Kafka 0.11.0.0 中正式引入的。

不论是哪个版本，Kafka 的消息层次都分为两层：消息集合（message set）以及消息（message）。一个消息集合中包含若干条日志项（record item），而日志项才是真正封装消息的地方。Kafka 底层的消息日志由一系列消息集合日志项组成。Kafka 通常不会直接操作具体的一条条消息，它总是在消息集合这个层面上进行写入操作。

那么社区引入 V2 版本的目的是什么呢？V2 版本主要是针对 V1 版本的一些弊端做了修正。

在 V1 版本中，每条消息都需要执行 CRC 校验。但有些情况下消息的 CRC 值是会发生变化的。比如在 Broker 端可能会对消息时间戳字段进行更新，那么重新计算之后的 CRC 值也会相应更新；再比如 Broker 端在执行消息格式转换时（主要是为了兼容老版本客户端程序），也会带来 CRC 值的变化。鉴于这些情况，再对每条消息都执行 CRC 校验就有点没必要了，不仅浪费空间还耽误 CPU 时间。

因此，在 V2 版本中，只对消息集合执行 CRC 校验。V2 版本还有一个和压缩息息相关的改进，就是保存压缩消息的方法发生了变化。之前 V1 版本中保存压缩消息的方法是把多条消息进行压缩然后保存到外层消息的消息体字段中；而 V2 版本的做法是对整个消息集合进行压缩。显然后者应该比前者有更好的压缩效果。

Kafka 的压缩流程

Kafka 的压缩流程，一言以概之——Producer 端压缩、Broker 端保持、Consumer 端解压缩。

压缩过程

在 Kafka 中，压缩可能发生在两个地方：生产者端和 Broker 端。

生产者程序中配置 compression.type 参数即表示启用指定类型的压缩算法。

【示例】开启 GZIP 的 Producer 对象

1
2
3
4
5
6
7
8
9

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 开启 GZIP 压缩
props.put("compression.type", "gzip");

Producer<String, String> producer = new KafkaProducer<>(props);

通常，Broker 从 Producer 端接收到消息后，不做任何处理。以下两种情况除外：

• 情况一：Broker 端指定了和 Producer 端不同的压缩算法。显然，应该尽量避免这种情况。

• 情况二：Broker 端发生了消息格式转换。所谓的消息格式转换，主要是为了兼容老版本的消费者程序。在一个生产环境中，Kafka 集群中同时保存多种版本的消息格式非常常见。为了兼容老版本的格式，Broker 端会对新版本消息执行向老版本格式的转换。这个过程中会涉及消息的解压缩和重新压缩。一般情况下这种消息格式转换对性能是有很大影响的，除了这里的压缩之外，它还让 Kafka 丧失了引以为豪的 Zero Copy 特性。

所谓零拷贝，说的是当数据在磁盘和网络进行传输时避免昂贵的内核态数据拷贝，从而实现快速的数据传输。因此如果 Kafka 享受不到这个特性的话，性能必然有所损失，所以尽量保证消息格式的统一吧，这样不仅可以避免不必要的解压缩 / 重新压缩，对提升其他方面的性能也大有裨益。

解压缩的过程

通常来说解压缩发生在消费者程序中，也就是说 Producer 发送压缩消息到 Broker 后，Broker 照单全收并原样保存起来。当 Consumer 程序请求这部分消息时，Broker 依然原样发送出去，当消息到达 Consumer 端后，由 Consumer 自行解压缩还原成之前的消息。

那么现在问题来了，Consumer 怎么知道这些消息是用何种压缩算法压缩的呢？其实答案就在消息中。Kafka 会将启用了哪种压缩算法封装进消息集合中，这样当 Consumer 读取到消息集合时，它自然就知道了这些消息使用的是哪种压缩算法。

压缩算法

在 Kafka 2.1.0 版本之前，Kafka 支持 3 种压缩算法：GZIP、Snappy 和 LZ4。从 2.1.0 开始，Kafka 正式支持 Zstandard 算法（简写为 zstd）。

在实际使用中，GZIP、Snappy、LZ4 甚至是 zstd 的表现各有千秋。但对于 Kafka 而言，它们的性能测试结果却出奇得一致，即在吞吐量方面：LZ4 > Snappy > zstd 和 GZIP；而在压缩比方面，zstd > LZ4 > GZIP > Snappy。

如果客户端机器 CPU 资源有很多富余，强烈建议开启 zstd 压缩，这样能极大地节省网络资源消耗。

何时启用压缩

何时启用压缩是比较合适的时机呢？

压缩是在 Producer 端完成的工作，那么启用压缩的一个条件就是 Producer 程序运行机器上的 CPU 资源要很充足。如果 Producer 运行机器本身 CPU 已经消耗殆尽了，那么启用消息压缩无疑是雪上加霜，只会适得其反。

如果环境中带宽资源有限，那么也建议开启压缩。

幂等性

什么是幂等性

幂等（idempotent、idempotence）是一个数学与计算机学概念，指的是：一个幂等操作的特点是其任意多次执行所产生的影响均与一次执行的影响相同。

Kafka Producer 的幂等性

在 Kafka 中，Producer 默认不是幂等性的，但我们可以创建幂等性 Producer。它其实是 0.11.0.0 版本引入的新功能。在此之前，Kafka 向分区发送数据时，可能会出现同一条消息被发送了多次，导致消息重复的情况。在 0.11 之后，指定 Producer 幂等性的方法很简单，仅需要设置一个参数即可，即 props.put(“enable.idempotence”, ture)，或 props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG， true)。

enable.idempotence 被设置成 true 后，Producer 自动升级成幂等性 Producer，其他所有的代码逻辑都不需要改变。Kafka 自动帮你做消息的去重。底层具体的原理很简单，就是经典的用空间去换时间的优化思路，即在 Broker 端多保存一些字段。当 Producer 发送了具有相同字段值的消息后，Broker 能够自动知晓这些消息已经重复了，于是可以在后台默默地把它们“丢弃”掉。当然，实际的实现原理并没有这么简单，但你大致可以这么理解。

我们必须要了解幂等性 Producer 的作用范围：

• 首先，**enable.idempotence 只能保证单分区上的幂等性**，即一个幂等性 Producer 能够保证某个主题的一个分区上不出现重复消息，它无法实现多个分区的幂等性。
• 其次，它只能实现单会话上的幂等性，不能实现跨会话的幂等性。这里的会话，你可以理解为 Producer 进程的一次运行。当你重启了 Producer 进程之后，这种幂等性保证就丧失了。

如果想实现多分区以及多会话上的消息无重复，应该怎么做呢？答案就是事务（transaction）或者依赖事务型 Producer。这也是幂等性 Producer 和事务型 Producer 的最大区别！

PID 和 Sequence Number

为了实现 Producer 的幂等性，Kafka 引入了 Producer ID（即 PID）和 Sequence Number。

• PID。每个新的 Producer 在初始化的时候会被分配一个唯一的 PID，这个 PID 对用户是不可见的。
• Sequence Numbler。对于每个 PID，该 Producer 发送数据的每个 <Topic, Partition> 都对应一个从 0 开始单调递增的 Sequence Number。

Broker 端在缓存中保存了这 seq number，对于接收的每条消息，如果其序号比 Broker 缓存中序号大于 1 则接受它，否则将其丢弃。这样就可以实现了消息重复提交了。但是，只能保证单个 Producer 对于同一个 <Topic, Partition> 的 Exactly Once 语义。不能保证同一个 Producer 一个 topic 不同的 partion 幂等。

实现幂等之后：

生成 PID 的流程

在执行创建事务时，如下：

1

Producer<String, String> producer = new KafkaProducer<String, String>(props);

会创建一个 Sender，并启动线程，执行如下 run 方法，在 maybeWaitForProducerId()中生成一个 producerId，如下：

1
2
3
4
5
6
7
8
9
10
11
12
13

====================================
类名：Sender
====================================

void run(long now) {
        if (transactionManager != null) {
            try {
                 ........
                if (!transactionManager.isTransactional()) {
                    // 为idempotent producer生成一个producer id
                    maybeWaitForProducerId();
                } else if (transactionManager.hasUnresolvedSequences() && !transactionManager.hasFatalError()) {
                   ........

幂等性的应用实例

（1）配置属性

需要设置：

• enable.idempotence，需要设置为 ture，此时就会默认把 acks 设置为 all，所以不需要再设置 acks 属性了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

// 指定生产者的配置
final Properties properties = new Properties();
properties.put("bootstrap.servers", "localhost:9092");
// 设置 key 的序列化器
properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 设置 value 的序列化器
properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// 开启幂等性
properties.put("enable.idempotence", true);
// 设置重试次数
properties.put("retries", 3);
//Reduce the no of requests less than 0
properties.put("linger.ms", 1);
// buffer.memory 控制生产者可用于缓冲的内存总量
properties.put("buffer.memory", 33554432);

// 使用配置初始化 Kafka 生产者
producer = new KafkaProducer<>(properties);

（2）发送消息

跟一般生产者一样，如下

1
2
3
4
5
6
7

public void produceIdempotMessage(String topic, String message) {
    // 创建Producer
    Producer producer = buildIdempotProducer();
    // 发送消息
    producer.send(new ProducerRecord<String, String>(topic, message));
    producer.flush();
}

此时，因为我们并没有配置 transaction.id 属性，所以不能使用事务相关 API，如下

1

producer.initTransactions();

否则会出现如下错误：

1
2
3
4

Exception in thread “main” java.lang.IllegalStateException: Transactional method invoked on a non-transactional producer.
    at org.apache.kafka.clients.producer.internals.TransactionManager.ensureTransactional(TransactionManager.java:777)
    at org.apache.kafka.clients.producer.internals.TransactionManager.initializeTransactions(TransactionManager.java:202)
    at org.apache.kafka.clients.producer.KafkaProducer.initTransactions(KafkaProducer.java:544)

Kafka 事务

Kafka 的事务概念是指一系列的生产者生产消息和消费者提交偏移量的操作在一个事务，或者说是是一个原子操作），同时成功或者失败。

消息可靠性保障，由低到高为：

• 最多一次（at most once）：消息可能会丢失，但绝不会被重复发送。
• 至少一次（at least once）：消息不会丢失，但有可能被重复发送。
• 精确一次（exactly once）：消息不会丢失，也不会被重复发送。

Kafka 支持事务功能主要是为了实现精确一次处理语义的，而精确一次处理是实现流处理的基石。

事务

Kafka 自 0.11 版本开始提供了对事务的支持，目前主要是在 read committed 隔离级别上做事情。它能保证多条消息原子性地写入到目标分区，同时也能保证 Consumer 只能看到事务成功提交的消息。

事务型 Producer

事务型 Producer 能够保证将消息原子性地写入到多个分区中。这批消息要么全部写入成功，要么全部失败。另外，事务型 Producer 也不惧进程的重启。Producer 重启回来后，Kafka 依然保证它们发送消息的精确一次处理。

事务属性实现前提是幂等性，即在配置事务属性 transaction.id 时，必须还得配置幂等性；但是幂等性是可以独立使用的，不需要依赖事务属性。

在事务属性之前先引入了生产者幂等性，它的作用为：

• 生产者多次发送消息可以封装成一个原子操作，要么都成功，要么失败。
• consumer-transform-producer 模式下，因为消费者提交偏移量出现问题，导致重复消费。需要将这个模式下消费者提交偏移量操作和生产者一系列生成消息的操作封装成一个原子操作。

消费者提交偏移量导致重复消费消息的场景：消费者在消费消息完成提交便宜量 o2 之前挂掉了（假设它最近提交的偏移量是 o1），此时执行再均衡时，其它消费者会重复消费消息(o1 到 o2 之间的消息）。

事务操作的 API

Producer 提供了 initTransactions, beginTransaction, sendOffsets, commitTransaction, abortTransaction 五个事务方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

/**
 * 初始化事务。需要注意的有：
 * 1、前提
 * 需要保证transation.id属性被配置。
 * 2、这个方法执行逻辑是：
 *   （1）Ensures any transactions initiated by previous instances of the producer with the same
 *      transactional.id are completed. If the previous instance had failed with a transaction in
 *      progress, it will be aborted. If the last transaction had begun completion,
 *      but not yet finished, this method awaits its completion.
 *    （2）Gets the internal producer id and epoch, used in all future transactional
 *      messages issued by the producer.
 *
 */
public void initTransactions();

/**
 * 开启事务
 */
public void beginTransaction() throws ProducerFencedException ;

/**
 * 为消费者提供的在事务内提交偏移量的操作
 */
public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
                                     String consumerGroupId) throws ProducerFencedException ;

/**
 * 提交事务
 */
public void commitTransaction() throws ProducerFencedException;

/**
 * 放弃事务，类似回滚事务的操作
 */
public void abortTransaction() throws ProducerFencedException ;

Kafka 事务相关配置

使用 kafka 的事务 api 时的一些注意事项：

• 需要消费者的自动模式设置为 false，并且不能子再手动的进行执行 consumer#commitSync 或者 consumer#commitAsyc
• 设置 Producer 端参数 transctional.id。最好为其设置一个有意义的名字。
• 和幂等性 Producer 一样，开启 enable.idempotence = true。如果配置了 transaction.id，则此时 enable.idempotence 会被设置为 true
• 消费者需要配置事务隔离级别 isolation.level。在 consume-trnasform-produce 模式下使用事务时，必须设置为 READ_COMMITTED。
  • read_uncommitted：这是默认值，表明 Consumer 能够读取到 Kafka 写入的任何消息，不论事务型 Producer 提交事务还是终止事务，其写入的消息都可以读取。很显然，如果你用了事务型 Producer，那么对应的 Consumer 就不要使用这个值。
  • read_committed：表明 Consumer 只会读取事务型 Producer 成功提交事务写入的消息。当然了，它也能看到非事务型 Producer 写入的所有消息。

Kafka 事务应用示例

只有生成操作

创建一个事务，在这个事务操作中，只有生成消息操作。代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

/**
 * 在一个事务只有生产消息操作
 */
public void onlyProduceInTransaction() {
    Producer producer = buildProducer();

    // 1.初始化事务
    producer.initTransactions();

    // 2.开启事务
    producer.beginTransaction();

    try {
        // 3.kafka写操作集合
        // 3.1 do业务逻辑

        // 3.2 发送消息
        producer.send(new ProducerRecord<String, String>("test", "transaction-data-1"));

        producer.send(new ProducerRecord<String, String>("test", "transaction-data-2"));
        // 3.3 do其他业务逻辑,还可以发送其他topic的消息。

        // 4.事务提交
        producer.commitTransaction();


    } catch (Exception e) {
        // 5.放弃事务
        producer.abortTransaction();
    }

}

创建生产者，代码如下,需要:

• 配置 transactional.id 属性
• 配置 enable.idempotence 属性

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

/**
 * 需要:
 * 1、设置transactional.id
 * 2、设置enable.idempotence
 * @return
 */
private Producer buildProducer() {

    // create instance for properties to access producer configs
    Properties props = new Properties();

    // bootstrap.servers是Kafka集群的IP地址。多个时,使用逗号隔开
    props.put("bootstrap.servers", "localhost:9092");

    // 设置事务id
    props.put("transactional.id", "first-transactional");

    // 设置幂等性
    props.put("enable.idempotence",true);

    //Set acknowledgements for producer requests.
    props.put("acks", "all");

    //If the request fails, the producer can automatically retry,
    props.put("retries", 1);

    //Specify buffer size in config,这里不进行设置这个属性,如果设置了,还需要执行producer.flush()来把缓存中消息发送出去
    //props.put("batch.size", 16384);

    //Reduce the no of requests less than 0
    props.put("linger.ms", 1);

    //The buffer.memory controls the total amount of memory available to the producer for buffering.
    props.put("buffer.memory", 33554432);

    // Kafka消息是以键值对的形式发送,需要设置key和value类型序列化器
    props.put("key.serializer",
            "org.apache.kafka.common.serialization.StringSerializer");

    props.put("value.serializer",
            "org.apache.kafka.common.serialization.StringSerializer");


    Producer<String, String> producer = new KafkaProducer<String, String>(props);

    return producer;
}

消费-生产并存（consume-transform-produce）

在一个事务中，既有生产消息操作又有消费消息操作，即常说的 Consume-tansform-produce 模式。如下实例代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

/**
 * 在一个事务内,即有生产消息又有消费消息
 */
public void consumeTransferProduce() {
    // 1.构建上产者
    Producer producer = buildProducer();
    // 2.初始化事务(生成productId),对于一个生产者,只能执行一次初始化事务操作
    producer.initTransactions();

    // 3.构建消费者和订阅主题
    Consumer consumer = buildConsumer();
    consumer.subscribe(Arrays.asList("test"));
    while (true) {
        // 4.开启事务
        producer.beginTransaction();

        // 5.1 接受消息
        ConsumerRecords<String, String> records = consumer.poll(500);

        try {
            // 5.2 do业务逻辑;
            System.out.println("customer Message---");
            Map<TopicPartition, OffsetAndMetadata> commits = Maps.newHashMap();
            for (ConsumerRecord<String, String> record : records) {
                // 5.2.1 读取消息,并处理消息。print the offset,key and value for the consumer records.
                System.out.printf("offset = %d, key = %s, value = %s\n",
                        record.offset(), record.key(), record.value());

                // 5.2.2 记录提交的偏移量
                commits.put(new TopicPartition(record.topic(), record.partition()),
                        new OffsetAndMetadata(record.offset()));


                // 6.生产新的消息。比如外卖订单状态的消息,如果订单成功,则需要发送跟商家结转消息或者派送员的提成消息
                producer.send(new ProducerRecord<String, String>("test", "data2"));
            }

            // 7.提交偏移量
            producer.sendOffsetsToTransaction(commits, "group0323");

            // 8.事务提交
            producer.commitTransaction();

        } catch (Exception e) {
            // 7.放弃事务
            producer.abortTransaction();
        }
    }
}

创建消费者代码，需要：

• 将配置中的自动提交属性（auto.commit）进行关闭
• 而且在代码里面也不能使用手动提交 commitSync( )或者 commitAsync( )
• 设置 isolation.level

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

/**
 * 需要:
 * 1、关闭自动提交 enable.auto.commit
 * 2、isolation.level为
 * @return
 */
public Consumer buildConsumer() {
    Properties props = new Properties();
    // bootstrap.servers是Kafka集群的IP地址。多个时,使用逗号隔开
    props.put("bootstrap.servers", "localhost:9092");
    // 消费者群组
    props.put("group.id", "group0323");
    // 设置隔离级别
    props.put("isolation.level","read_committed");
    // 关闭自动提交
    props.put("enable.auto.commit", "false");
    props.put("session.timeout.ms", "30000");
    props.put("key.deserializer",
            "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer",
            "org.apache.kafka.common.serialization.StringDeserializer");
    KafkaConsumer<String, String> consumer = new KafkaConsumer
            <String, String>(props);

    return consumer;
}

只有消费操作

创建一个事务，在这个事务操作中，只有生成消息操作，如下代码。这种操作其实没有什么意义，跟使用手动提交效果一样，无法保证消费消息操作和提交偏移量操作在一个事务。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

/**
 * 在一个事务只有消息操作
 */
public void onlyConsumeInTransaction() {
    Producer producer = buildProducer();

    // 1.初始化事务
    producer.initTransactions();

    // 2.开启事务
    producer.beginTransaction();

    // 3.kafka读消息的操作集合
    Consumer consumer = buildConsumer();
    while (true) {
        // 3.1 接受消息
        ConsumerRecords<String, String> records = consumer.poll(500);

        try {
            // 3.2 do业务逻辑;
            System.out.println("customer Message---");
            Map<TopicPartition, OffsetAndMetadata> commits = Maps.newHashMap();
            for (ConsumerRecord<String, String> record : records) {
                // 3.2.1 处理消息 print the offset,key and value for the consumer records.
                System.out.printf("offset = %d, key = %s, value = %s\n",
                        record.offset(), record.key(), record.value());

                // 3.2.2 记录提交偏移量
                commits.put(new TopicPartition(record.topic(), record.partition()),
                        new OffsetAndMetadata(record.offset()));
            }

            // 4.提交偏移量
            producer.sendOffsetsToTransaction(commits, "group0323");

            // 5.事务提交
            producer.commitTransaction();

        } catch (Exception e) {
            // 6.放弃事务
            producer.abortTransaction();
        }
    }

}

生产者的配置

更详尽的生产者配置可以参考：Kafka 生产者官方配置说明

以下为生产者主要配置参数清单：

• acks：指定了必须有多少个分区副本收到消息，生产者才会认为消息写入是成功的。默认为 acks=1
  • acks=0 如果设置为 0，则 Producer 不会等待服务器的反馈。该消息会被立刻添加到 socket buffer 中并认为已经发送完成。在这种情况下，服务器是否收到请求是没法保证的，并且参数retries也不会生效（因为客户端无法获得失败信息）。每个记录返回的 offset 总是被设置为-1。
  • acks=1 如果设置为 1，leader 节点会将记录写入本地日志，并且在所有 follower 节点反馈之前就先确认成功。在这种情况下，如果 leader 节点在接收记录之后，并且在 follower 节点复制数据完成之前产生错误，则这条记录会丢失。
  • acks=all 如果设置为 all，这就意味着 leader 节点会等待所有同步中的副本确认之后再确认这条记录是否发送完成。只要至少有一个同步副本存在，记录就不会丢失。这种方式是对请求传递的最有效保证。acks=-1 与 acks=all 是等效的。
• buffer.memory：用来设置 Producer 缓冲区大小。
• compression.type：Producer 生成数据时可使用的压缩类型。默认值是 none(即不压缩)。可配置的压缩类型包括：none、gzip、snappy 、lz4 或 zstd。压缩是针对批处理的所有数据，所以批处理的效果也会影响压缩比（更多的批处理意味着更好的压缩）。
• retries：用来设置发送失败的重试次数。
• batch.size：用来设置一个批次可占用的内存大小。
• linger.ms：用来设置 Producer 在发送批次前的等待时间。
• client.id：Kafka 服务器用它来识别消息源，可以是任意字符串。
• max.in.flight.requests.per.connection：用来设置 Producer 在收到服务器响应前可以发送多少个消息。
• timeout.ms：用来设置 Broker 等待同步副本返回消息确认的时间，与 acks 的配置相匹配。
• request.timeout.ms：Producer 在发送数据时等待服务器返回响应的时间。
• metadata.fetch.timeout.ms：Producer 在获取元数据时（如：分区的 Leader 是谁）等待服务器返回响应的时间。
• max.block.ms：该配置控制 KafkaProducer.send() 和KafkaProducer.partitionsFor() 允许被阻塞的时长。这些方法可能因为缓冲区满了或者元数据不可用而被阻塞。用户提供的序列化程序或分区程序的阻塞将不会被计算到这个超时。
• max.request.size：请求的最大字节数。
• receieve.buffer.bytes：TCP 接收缓冲区的大小。
• send.buffer.bytes：TCP 发送缓冲区的大小。

参考资料

• 官方
  • Kafka 官网
  • Kafka Github
  • Kafka 官方文档
  • Kafka Confluent 官网
  • Kafka Jira
• 书籍
  • 《Kafka 权威指南》
  • 《深入理解 Kafka：核心设计与实践原理》
  • 《Kafka 技术内幕》
• 教程
  • Kafka 中文文档
  • Kafka 核心技术与实战
• 文章
  • Kafak(04) Kafka 生产者事务和幂等

深入理解 Java String 类型

String 类型可能是 Java 中应用最频繁的引用类型，但它的性能问题却常常被忽略。高效的使用字符串，可以提升系统的整体性能。当然，要做到高效使用字符串，需要深入了解其特性。

String 的不可变性

我们先来看下 String 的定义：

1
2
3
4

public final class String
    implements java.io.Serializable, Comparable<String>, CharSequence {
    /** The value is used for character storage. */
    private final char value[];

String 类被 final 关键字修饰，表示不可继承 String 类。

String 类的数据存储于 char[] 数组，这个数组被 final 关键字修饰，表示 String 对象不可被更改。

为什么 Java 要这样设计？

（1）保证 String 对象安全性。避免 String 被篡改。

（2）保证 hash 值不会频繁变更。

（3）可以实现字符串常量池。通常有两种创建字符串对象的方式，一种是通过字符串常量的方式创建，如 String str="abc"; 另一种是字符串变量通过 new 形式的创建，如 String str = new String("abc")。

使用第一种方式创建字符串对象时，JVM 首先会检查该对象是否在字符串常量池中，如果在，就返回该对象引用，否则新的字符串将在常量池中被创建。这种方式可以减少同一个值的字符串对象的重复创建，节约内存。

String str = new String("abc") 这种方式，首先在编译类文件时，"abc" 常量字符串将会放入到常量结构中，在类加载时，"abc" 将会在常量池中创建；其次，在调用 new 时，JVM 命令将会调用 String 的构造函数，同时引用常量池中的 "abc" 字符串，在堆内存中创建一个 String 对象；最后，str 将引用 String 对象。

String 的性能考量

字符串拼接

字符串常量的拼接，编译器会将其优化为一个常量字符串。

【示例】字符串常量拼接

1
2
3
4
5
6

public static void main(String[] args) {
    // 本行代码在 class 文件中，会被编译器直接优化为：
    // String str = "abc";
    String str = "a" + "b" + "c";
    System.out.println("str = " + str);
}

字符串变量的拼接，编译器会优化成 StringBuilder 的方式。

【示例】字符串变量的拼接

1
2
3
4
5
6
7
8

public static void main(String[] args) {
    String str = "";
    for(int i=0; i<1000; i++) {
        // 本行代码会被编译器优化为：
        // str = (new StringBuilder(String.valueOf(str))).append(i).toString();
        str = str + i;
    }
}

但是，每次循环都会生成一个新的 StringBuilder 实例，同样也会降低系统的性能。

字符串拼接的正确方案：

• 如果需要使用字符串拼接，应该优先考虑 StringBuilder 的 append 方法替代使用 + 号。
• 如果在并发编程中，String 对象的拼接涉及到线程安全，可以使用 StringBuffer。但是要注意，由于 StringBuffer 是线程安全的，涉及到锁竞争，所以从性能上来说，要比 StringBuilder 差一些。

字符串分割

String 的 split() 方法使用正则表达式实现其强大的分割功能。而正则表达式的性能是非常不稳定的，使用不恰当会引起回溯问题，很可能导致 CPU 居高不下。

所以，应该慎重使用 split() 方法，可以考虑用 String.indexOf() 方法代替 split() 方法完成字符串的分割。如果实在无法满足需求，你就在使用 Split() 方法时，对回溯问题加以重视就可以了。

String.intern

在每次赋值的时候使用 String 的 intern 方法，如果常量池中有相同值，就会重复使用该对象，返回对象引用，这样一开始的对象就可以被回收掉。

在字符串常量中，默认会将对象放入常量池；在字符串变量中，对象是会创建在堆内存中，同时也会在常量池中创建一个字符串对象，复制到堆内存对象中，并返回堆内存对象引用。

如果调用 intern 方法，会去查看字符串常量池中是否有等于该对象的字符串，如果没有，就在常量池中新增该对象，并返回该对象引用；如果有，就返回常量池中的字符串引用。堆内存中原有的对象由于没有引用指向它，将会通过垃圾回收器回收。

【示例】

1
2
3
4
5
6
7
8
9
10

public class SharedLocation {

	private String city;
	private String region;
	private String countryCode;
}

SharedLocation sharedLocation = new SharedLocation();
sharedLocation.setCity(messageInfo.getCity().intern());		sharedLocation.setCountryCode(messageInfo.getRegion().intern());
sharedLocation.setRegion(messageInfo.getCountryCode().intern());

使用 intern 方法需要注意：一定要结合实际场景。因为常量池的实现是类似于一个 HashTable 的实现方式，HashTable 存储的数据越大，遍历的时间复杂度就会增加。如果数据过大，会增加整个字符串常量池的负担。

String、StringBuffer、StringBuilder 有什么区别

String 是 Java 语言非常基础和重要的类，提供了构造和管理字符串的各种基本逻辑。它是典型的 Immutable 类，被声明成为 final class，所有属性也都是 final 的。也由于它的不可变性，类似拼接、裁剪字符串等动作，都会产生新的 String 对象。由于字符串操作的普遍性，所以相关操作的效率往往对应用性能有明显影响。

StringBuffer 是为解决上面提到拼接产生太多中间对象的问题而提供的一个类，我们可以用 append 或者 add 方法，把字符串添加到已有序列的末尾或者指定位置。StringBuffer 是一个线程安全的可修改字符序列。StringBuffer 的线程安全是通过在各种修改数据的方法上用 synchronized 关键字修饰实现的。

StringBuilder 是 Java 1.5 中新增的，在能力上和 StringBuffer 没有本质区别，但是它去掉了线程安全的部分，有效减小了开销，是绝大部分情况下进行字符串拼接的首选。

StringBuffer 和 StringBuilder 底层都是利用可修改的（char，JDK 9 以后是 byte）数组，二者都继承了 AbstractStringBuilder，里面包含了基本操作，区别仅在于最终的方法是否加了 synchronized。构建时初始字符串长度加 16（这意味着，如果没有构建对象时输入最初的字符串，那么初始值就是 16）。我们如果确定拼接会发生非常多次，而且大概是可预计的，那么就可以指定合适的大小，避免很多次扩容的开销。扩容会产生多重开销，因为要抛弃原有数组，创建新的（可以简单认为是倍数）数组，还要进行 arraycopy。

**除非有线程安全的需要，不然一般都使用 StringBuilder**。

参考资料

• 《Java 编程思想（Thinking in java）》
• 《Java 核心技术 卷 I 基础知识》
• 极客时间教程 - Java 性能调优实战
• 极客时间教程 - Java 核心技术面试精讲
• Java 基本数据类型和引用类型
• 深入剖析 Java 中的装箱和拆箱

Java 正则从入门到精通

关键词：Pattern、Matcher、捕获与非捕获、反向引用、零宽断言、贪婪与懒惰、元字符、DFA、NFA

正则简介

正则表达式是什么

正则表达式（Regular Expression）是一个用正则符号写出的公式，程序对这个公式进行语法分析，建立一个语法分析树，再根据这个分析树结合正则表达式的引擎生成执行程序（这个执行程序我们把它称作状态机，也叫状态自动机），用于字符匹配。

如何学习正则

正则表达式是一个强大的文本匹配工具，但是它的规则很复杂，理解起来较为困难，容易让人望而生畏。

刚接触正则时，我看了一堆正则的语义说明，但是仍然不明所以。后来，我多接触一些正则的应用实例，渐渐有了感觉，再结合语义说明，终有领悟。我觉得正则表达式和武侠修练武功差不多，应该先练招式，再练心法。如果一开始就直接看正则的规则，保证你会懵逼。当你熟悉基本招式（正则基本使用案例）后，也该修炼修炼心法（正则语法）了。真正的高手不能只靠死记硬背那么几招把式。就像张三丰教张无忌太极拳一样，领悟心法，融会贯通，少侠你就可以无招胜有招，成为传说中的绝世高手。

以上闲话可归纳为一句：学习正则应该从实例去理解规则。

正则工具类

JDK 中的 java.util.regex 包提供了对正则表达式的支持。

java.util.regex 有三个核心类：

• Pattern 类：Pattern 是一个正则表达式的编译表示。
• Matcher 类：Matcher 是对输入字符串进行解释和匹配操作的引擎。
• PatternSyntaxException：PatternSyntaxException 是一个非强制异常类，它表示一个正则表达式模式中的语法错误。

注：需要格外注意一点，在 Java 中使用反斜杠”\“时必须写成 "\\"。所以本文的代码出现形如 String regex = "\\$\\{.*?\\}" 其实就是 \$\{.\*?\}。

Pattern 类

Pattern类没有公共构造方法。要创建一个Pattern对象，你必须首先调用其静态方法compile，加载正则规则字符串，然后返回一个 Pattern 对象。

与Pattern类一样，Matcher类也没有公共构造方法。你需要调用Pattern对象的matcher方法来获得一个Matcher对象。

【示例】Pattern 和 Matcher 的初始化

1
2

Pattern p = Pattern.compile(regex);
Matcher m = p.matcher(content);

Matcher 类

Matcher 类可以说是 java.util.regex 中的核心类，它有三类功能：校验、查找、替换。

校验

为了校验文本是否与正则规则匹配，Matcher 提供了以下几个返回值为 boolean 的方法。

如果你傻傻分不清上面的查找方法有什么区别，那么下面一个例子就可以让你秒懂。

【示例】lookingAt、find、matches

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

public static void main(String[] args) {
	checkLookingAt("hello", "helloworld");
	checkLookingAt("world", "helloworld");

	checkFind("hello", "helloworld");
	checkFind("world", "helloworld");

	checkMatches("hello", "helloworld");
	checkMatches("world", "helloworld");
	checkMatches("helloworld", "helloworld");
}

private static void checkLookingAt(String regex, String content) {
	Pattern p = Pattern.compile(regex);
	Matcher m = p.matcher(content);
	if (m.lookingAt()) {
		System.out.println(content + "\tlookingAt： " + regex);
	} else {
		System.out.println(content + "\tnot lookingAt： " + regex);
	}
}

private static void checkFind(String regex, String content) {
	Pattern p = Pattern.compile(regex);
	Matcher m = p.matcher(content);
	if (m.find()) {
		System.out.println(content + "\tfind： " + regex);
	} else {
		System.out.println(content + "\tnot find： " + regex);
	}
}

private static void checkMatches(String regex, String content) {
	Pattern p = Pattern.compile(regex);
	Matcher m = p.matcher(content);
	if (m.matches()) {
		System.out.println(content + "\tmatches： " + regex);
	} else {
		System.out.println(content + "\tnot matches： " + regex);
	}
}

输出：

1
2
3
4
5
6
7

helloworld	lookingAt： hello
helloworld	not lookingAt： world
helloworld	find： hello
helloworld	find： world
helloworld	not matches： hello
helloworld	not matches： world
helloworld	matches： helloworld

说明

regex = "world" 表示的正则规则是以 world 开头的字符串，regex = "hello" 和 regex = "helloworld" 也是同理。

• lookingAt方法从头部开始，检查 content 字符串是否有子字符串于正则规则匹配。
• find方法检查 content 字符串是否有子字符串于正则规则匹配，不管字符串所在位置。
• matches方法检查 content 字符串整体是否与正则规则匹配。

查找

为了查找文本匹配正则规则的位置，Matcher提供了以下方法：

【示例】使用 start()、end()、group() 查找所有匹配正则条件的子序列

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

public static void main(String[] args) {
	final String regex = "world";
	final String content = "helloworld helloworld";
	Pattern p = Pattern.compile(regex);
	Matcher m = p.matcher(content);
	System.out.println("content: " + content);

	int i = 0;
	while (m.find()) {
		i++;
		System.out.println("[" + i + "th] found");
		System.out.print("start: " + m.start() + ", ");
		System.out.print("end: " + m.end() + ", ");
		System.out.print("group: " + m.group() + "\n");
	}
}

输出

1
2
3
4
5

content: helloworld helloworld
[1th] found
start: 5, end: 10, group: world
[2th] found
start: 16, end: 21, group: world

说明

例子很直白，不言自明了吧。

替换

替换方法是替换输入字符串里文本的方法：

【示例】replaceFirst 和 replaceAll

1
2
3
4
5
6
7
8
9
10
11
12

public static void main(String[] args) {
	String regex = "can";
	String replace = "can not";
	String content = "I can because I think I can.";

	Pattern p = Pattern.compile(regex);
	Matcher m = p.matcher(content);

	System.out.println("content: " + content);
	System.out.println("replaceFirst: " + m.replaceFirst(replace));
	System.out.println("replaceAll: " + m.replaceAll(replace));
}

输出

1
2
3

content: I can because I think I can.
replaceFirst: I can not because I think I can.
replaceAll: I can not because I think I can not.

说明

replaceFirst：替换第一个匹配正则规则的子序列。

replaceAll：替换所有匹配正则规则的子序列。

【示例】appendReplacement、appendTail 和 replaceAll

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

public static void main(String[] args) {
	String regex = "can";
	String replace = "can not";
	String content = "I can because I think I can.";
	StringBuffer sb = new StringBuffer();
	StringBuffer sb2 = new StringBuffer();

	System.out.println("content: " + content);
	Pattern p = Pattern.compile(regex);
	Matcher m = p.matcher(content);
	while (m.find()) {
		m.appendReplacement(sb, replace);
	}
	System.out.println("appendReplacement: " + sb);
	m.appendTail(sb);
	System.out.println("appendTail: " + sb);
}

输出

1
2
3

content: I can because I think I can.
appendReplacement: I can not because I think I can not
appendTail: I can not because I think I can not.

说明

从输出结果可以看出，appendReplacement和appendTail方法组合起来用，功能和replaceAll是一样的。

如果你查看replaceAll的源码，会发现其内部就是使用appendReplacement和appendTail方法组合来实现的。

【示例】quoteReplacement 和 replaceAll，解决特殊字符替换问题

1
2
3
4
5
6
7
8
9
10
11
12

public static void main(String[] args) {
	String regex = "\\$\\{.*?\\}";
	String replace = "${product}";
	String content = "product is ${productName}.";

	Pattern p = Pattern.compile(regex);
	Matcher m = p.matcher(content);
	String replaceAll = m.replaceAll(replace);

	System.out.println("content: " + content);
	System.out.println("replaceAll: " + replaceAll);
}

输出

1
2
3
4
5
6
7
8
9
10

Exception in thread "main" java.lang.IllegalArgumentException: No group with name {product}
	at java.util.regex.Matcher.appendReplacement(Matcher.java:849)
	at java.util.regex.Matcher.replaceAll(Matcher.java:955)
	at org.zp.notes.javase.regex.RegexDemo.wrongMethod(RegexDemo.java:42)
	at org.zp.notes.javase.regex.RegexDemo.main(RegexDemo.java:18)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at com.intellij.rt.execution.application.AppMain.main(AppMain.java:147)

说明

String regex = "\\$\\{.*?\\}";表示匹配类似${name}这样的字符串。由于$、{ 、}都是特殊字符，需要用反义字符\来修饰才能被当做一个字符串字符来处理。

上面的例子是想将 ${productName} 替换为 ${product} ，然而replaceAll方法却将传入的字符串中的$当做特殊字符来处理了。结果产生异常。

如何解决这个问题?

JDK1.5 引入了quoteReplacement方法。它可以用来转换特殊字符。其实源码非常简单，就是判断字符串中如果有\或$，就为它加一个转义字符\