发消息就用消息队列——kafka、rocketMQ(一)_消息队列的应用场景rockmq,kfaka

前言

可能都听说过“消息队列”，这个都是为了帮助我们可以处理数据或者暂时保存数据用的。
kafka主要是一个分布式流媒体平台，主要是用来处理消息的，追求的是高吞吐量，一开始的目的就是用于日志收集和传输，适合产生大量数据的数据处理业务。比如我们使用的日志收集的时候（ELK）
rocketMQ是用来保存数据的消息队列，用来应用解耦或者流量削峰，比如说淘宝双11的时候用来业务削峰，当有大量交易涌入的时候，后端可能没办法立马处理，可以利用它进行缓存一下。

kafka介绍

知识点介绍：
topic： Kafka将消息分门别类，每一类的消息称之为一个主题
producer 发布消息的对象称之为生产者
consumer 订阅消息并处理发布的消息的对象称之为消费者
broker 已发布的消息保存在一组服务器中，称之为Kafka集群。集群中的每一个服务器都是一个代理（Broker）。 消费者可以订阅一个或多个主题（topic），并从Broker拉数据，从而消费这些已发布的消息。
kafka安装
在使用kafka之前，要先有她所需要的环境，首先Java环境，其次需要zookeeper，所以我们要装一个zookeeper，请查看zookeeper安装

这些准备好以后，开始安装kafka

上传到服务器，解压

tar -zxvf kafka_2.13-3.0.0.tgz
1

修改参数，修改config下面得server.properties

启动kafka，注意：在启动kafka之前必须先启动zookeeper，因为之前说过了kafka是依赖于zookeeper的。

bin/kafka-server-start.sh config/server.properties #启动kafka

1
2

生产者详解
原理：

发送类型：

• 发送并忘记
  把消息发送给服务器，并不关心它是否正常到达，大多数情况下，消息会正常达到，因为kafka是高可用的，而生产者会自动尝试重新发送，使用这种方式可能会丢失一部分数据，因为kafka不会返回确认信息的。
• 同步发送
  使用send()发送，他会返回一个Future对象，调用get()方法进行等待，就可以知道消息是否发送成功

try {
   //发送消息
   RecordMetadata recordMetadata = producer.send(record).get();
   System.out.println(recordMetadata.offset());
} catch (Exception e) {
   e.printStackTrace();
}
1
2
3
4
5
6
7

如果服务器返回错误，get()方法会抛出异常，如果没有发生错误，我们就会得到一个RecordMetadata对象，可以用它来获取消息的偏移量.

• 异步发送
  调用send()方法，并指定一个回调函数，服务器在返回响应时调用函数。

	try {
          
            //异步发送消息
            producer.send(record, new Callback() {
                //当消息发生成功调用的方法
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        if (e!=null){
                            e.printStackTrace();
                        }
                    System.out.println(recordMetadata.offset());
                }
            });

        } catch (Exception e) {
            e.printStackTrace();
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

如果kafka返回一个错误，onCompletion()方法会抛出一个非空（non null）异常，可以根据实际情况处理，比如记录错误日志，或者把消息写入“错误消息”文件中，方便后期进行分析。

参数详解：

• acks
  生产者的消息发送确认机制
  acks=0：就是不要得到什么反馈，很可能会丢失消息，所以它可以以网络能够支持的最大速度发送消息，从而达到最大的吞吐量。
  acks=1：就是首领节点接收消息不管成功失败，生产者都会接收到一条信息，如果接收到一个失败消息，为了避免数据错误，生产者就会重新发送一次消息。
  acks=all:当参数的所有节点都成功接收到消息时，生产者才会收到一个成功的消息，这种模式是最安全的，它可以保证不止一个服务器收到消息，就算有服务器发生崩溃，整个集群仍然可以运行。不过他的延迟比acks=1时更高。
  reteries:这个参数的作用就是设置重试的次数，如果超过了这次的重试次数就会返回错误。默认情况下，生产者会在每次重试之间等待100ms

消费者详解
原理

参数详解：

• enable.auto.commit
  该属性指定了消费者是否自动提交偏移量，默认值是true。为了尽量避免出现重复数据和数据丢失，可以把它设置为false，由自己控制何时提交偏移量。如果把它设置为true,还可以通过配置auto.commit.interval.ms属性来控制提交的频率。
• 提交和偏移量
  所谓偏移量就是指的消息在分区中的位置，当消费者发生崩溃或者有新的消费者加入群组，就会触发再均衡，完成均衡以后，这个时候需要消费者去去读取每一个偏移量，然后根据最后一次提交的偏移量，然后找到执行位置继续处理。
  如果提交偏移量小于客户端处理的最后一个消息的偏移量，那么处于两个偏移量之间的消息就会被重复处理。
  

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

• 自动提交偏移量
  当enable.auto.commit被设置为true，提交方式就是让消费者自动提交偏移量，每隔5秒消费者会自动把从poll()方法接收的最大偏移量提交上去。提交时间间隔有auto.commot.interval.ms控制，默认值是5秒。
• 提交当前偏移量（同步提交）
  把enable.auto.commit设置为false,让应用程序决定何时提交偏移量。使用commitSync()提交偏移量，commitSync()将会提交poll返回的最新的偏移量，所以在处理完所有记录后要确保调用了commitSync()方法。否则还是会有消息丢失的风险。

properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"flase");//设置不自动提交偏移量

while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record.value());
                System.out.println(record.key());

                try{
                    //手动提交偏移量
                    consumer.commitSync();//同步提交
                }catch (CommitFailedException e){
                    e.printStackTrace();
                    System.out.println("记录错误信息为："+e);
                }

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

• 异步提交
  手动提交有一个缺点，那就是当发起提交调用时应用会阻塞。当然我们可以减少手动提交的频率，但这个会增加消息重复的概率（和自动提交一样）。另外一个解决办法是，使用异步提交的API。

properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"flase");//设置不自动提交偏移量


while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record.value());
                System.out.println(record.key());
            }
            
            //异步手动提交偏移量
            consumer.commitAsync(new OffsetCommitCallback() {
                @Override
                public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
                    if(e!=null){
                        System.out.println("记录错误的提交偏移量："+ map+",异常信息"+e);
                    }
                }
            });

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

• 异步与同步组合提交
  异步提交也有个缺点，那就是如果服务器返回提交失败，异步提交不会进行重试。相比较起来，同步提交会进行重试直到成功或者最后抛出异常给应用。异步提交没有实现重试是因为，如果同时存在多个异步提交，进行重试可能会导致位移覆盖。

properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"flase");//设置不自动提交偏移量


try {
            while (true){
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println(record.value());
                    System.out.println(record.key());
                }
                //异步发送
                consumer.commitAsync();
            }
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("记录错误信息："+e);
        }finally {
            try {
                //同步发送
                consumer.commitSync();
            }finally {
                consumer.close();
            }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

上面简单介绍了kafka的生产者与消费者，下面就是正式的使用。

springboot与kafka整合
导入依赖

        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.2.7.RELEASE</version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.kafka</groupId>
                    <artifactId>kafka-clients</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-streams</artifactId>
            <version>2.0.1</version>
            <exclusions>
                <exclusion>
                    <artifactId>connect-json</artifactId>
                    <groupId>org.apache.kafka</groupId>
                </exclusion>
                <exclusion>
                    <groupId>org.apache.kafka</groupId>
                    <artifactId>kafka-clients</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>2.0.1</version>
        </dependency>
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

编写配置文件

server:
  port: 9991
spring:
  application:
    name: kafka-demo
  kafka:
    bootstrap-servers: 10.199.12.155:9092
    producer:
      retries: 0
      batch-size: 16368
      buffer-memory: 33554432
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: test-hello-group
      auto-offset-reset: earliest
      enable-auto-commit: true
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

编写生产者

@RestController
public class HelloController {

    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    @GetMapping("/hello")
    public String hello(){
        //第一个参数：topics  
        //第二个参数：消息内容
        kafkaTemplate.send("hello-itcast","hello springboot kafka");
        return "ok";
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

编写消费者

@Component
public class HelloListener {

    @KafkaListener(topics = {"hello-itcast"})
    public void receiverMessage(ConsumerRecord<?,?> record){
        Optional<? extends ConsumerRecord<?, ?>> optional = Optional.ofNullable(record);
        if(optional.isPresent()){
            Object value = record.value();
            System.out.println(value);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12

然后启动项目即可。

注意点:

因为springboot整合kafka以后，由于使用的序列化器是StringSerializer，这个时候 生产者与消费者之间如果传递对象的时候，有两种方式：
第一：自定义类序列化器，但是这种通用性不强，没传递一个类型的类都要定义一个，很不方便。
第二：因为我们使用的是字符串序列化器，所以，我们可以选择先把要传递的对象转化为json字符串，接收完消息以后再将其转化为对象即可。所以还需再引入一个jar来做这个转化

 <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.58</version>
        </dependency>
1
2
3
4
5

传递对象的方式

@GetMapping("/hello2")
    public String hello2(){

        User user = new User();
        user.setUsername("zhangsan");
        user.setAge(18);

        //第一个参数：topics
        //第二个参数：消息内容
        kafkaTemplate.send("hello-itcast", JSON.toJSONString(user));
        return "ok";
}
1
2
3
4
5
6
7
8
9
10
11
12

就如上面的例子，先转化为字符串，在发送

@Component
public class UserListener {

    @KafkaListener(topics = {"user-itcast"})
    public void receiverMessage(ConsumerRecord<?,?> record){
        Optional<? extends ConsumerRecord<?, ?>> optional = Optional.ofNullable(record);
        if(optional.isPresent()){
            Object value = record.value();
            User user = JSON.parseObject(String.valueOf(value), User.class);
            System.out.println(user);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

然后接收到后，再利用json将其转化为对应的对象即可。

rocketMQ介绍