壁纸壁布施工航飞曹 發表於 2022-11-14 23:13:00

云服务器(Linux)安装部署Kafka

<h1 id="云服务器linux安装部署kafka">云服务器(Linux)安装部署Kafka</h1>
<h2 id="前期准备">前期准备</h2>
<p>kafka的安装需要依赖于jdk，需要在服务器上提前安装好该环境，这里使用用jdk1.8。</p>
<h2 id="下载安装包">下载安装包</h2>
<p>官网地址：</p>
<p>较新的版本已自带Zookeeper，无需额外下载。这里使用3.2.0做演示。</p>
<p><img src="https://img2022.cnblogs.com/blog/2286325/202211/2286325-20221114231053741-188446239.png"></p>
<p>注意要下载<strong>Binary downloads</strong>标签下的tgz包，<strong>Source download</strong>标签下的包为源码。无法直接运行，需要编译。</p>
<h2 id="上载安装包到云服务器">上载安装包到云服务器</h2>
<p>使用ssh连接工具将kafka_2.12-3.2.0.tgz这个包上传到云服务器上的一个目录。</p>
<p><img src="https://img2022.cnblogs.com/blog/2286325/202211/2286325-20221114231110516-1489918447.png"></p>
<p>打开命令行，进入到放有压缩包的目录，执行</p>
<pre><code class="language-sh">tar -zxvf kafka_2.12-3.2.0.tgz
</code></pre>
<h2 id="配置kafka">配置kafka</h2>
<p>然后使用cd命令进入到/kafka_2.12-3.2.0/config/下，使用</p>
<pre><code class="language-sh">vi server.properties
</code></pre>
<p>编辑配置文件。</p>
<p><img src="https://img2022.cnblogs.com/blog/2286325/202211/2286325-20221114231145026-870644779.png"></p>
<p>删除listeners和advertised前方的#号，改成如下配置:</p>
<pre><code class="language-properties">listeners=PLAINTEXT://云服务器内网ip:9092（本地访问用本地ip）
# 如果要提供外网访问则必须配置此项
advertised.listeners=PLAINTEXT://云服务器公网ip:9092（若要远程访问需配置此项为云服务器的公网ip）
# zookeeper连接地址，集群配置格式为ip:port,ip:port,ip:port
zookeeper.connect=云服务器公网ip:2181
</code></pre>
<h2 id="开放云服务器端口">开放云服务器端口</h2>
<p>在云服务器控制台内进入安全组页面，添加两条新的入站规则，tcp/9092和tcp/2181</p>
<h2 id="开放linux防火墙端口">开放linux防火墙端口</h2>
<p>先查看使用的防火墙类型iptables/firewalld</p>
<p>iptables操作命令</p>
<pre><code class="language-sh">1.打开/关闭/重启防火墙

开启防火墙(重启后永久生效)：chkconfig iptables on

关闭防火墙(重启后永久生效)：chkconfig iptables off

开启防火墙(即时生效，重启后失效)：service iptables start

关闭防火墙(即时生效，重启后失效)：service iptables stop

重启防火墙:service iptables restartd

2.查看打开的端口

/etc/init.d/iptables status
3.开启端口

iptables -A INPUT -p tcp --dport 8080 -j ACCEPT
4.保存并重启防火墙
/etc/rc.d/init.d/iptables save
/etc/init.d/iptables restart
</code></pre>
<p>Centos7默认安装了firewalld，如果没有安装的话，可以使用 yum install firewalld firewalld-config进行安装。</p>
<p>操作指令如下:</p>
<pre><code class="language-sh">1.启动防火墙

systemctl start firewalld
2.禁用防火墙

systemctl stop firewalld
3.设置开机启动

systemctl enable firewalld
4.停止并禁用开机启动

sytemctl disable firewalld
5.重启防火墙

firewall-cmd --reload

6.查看状态

systemctl status firewalld或者 firewall-cmd --state
7.在指定区域打开端口（记得重启防火墙）

firewall-cmd --zone=public --add-port=80/tcp(永久生效再加上 --permanent)
</code></pre>
<p>打开tcp/9092和tcp/2181这两个端口后，重启防火墙，并查看开放的端口确实生效。</p>
<h2 id="启动kafka服务">启动kafka服务</h2>
<p>cd命令进入kafka_2.12-3.2.0目录下，执行</p>
<pre><code>bin/zookeeper-server-start.sh config/zookeeper.properties
</code></pre>
<p>启动zookeeper,不加-daemon方便排除启动错误，新建一个shell窗口，进入该目录再执行</p>
<pre><code class="language-sh">bin/kafka-server-start.sh config/server.properties
</code></pre>
<p>启动kafka,若打印日志未报错，若未出现error日志，说明启动成功。</p>
<h2 id="测试单机连通性">测试单机连通性</h2>
<pre><code>查询kafka下所有的topic
bin/kafka-topics.sh --list --zookeeper ip:port
因为kafka使用zookeeper作为配置中心，一些topic信息需要查询该kafka对应的zookeeper
创建topic
bin/kafka-topics.sh --create --zookeeper ip:port --replication-factor 1 --partitions 1 --topic test
开启生产者
bin/kafka-console-producer.sh --broker-list cos100:9092 --topic test
开启消费者
bin/kafka-console-consumer.sh --bootstrap-server cos100:9092 --topic test


</code></pre>
<h2 id="springboot连接kafak">Springboot连接kafak</h2>
<p>在pom.xml文件中引入kafka依赖</p>
<pre><code class="language-xml">                &lt;dependency&gt;
            &lt;groupId&gt;org.springframework.kafka&lt;/groupId&gt;
            &lt;artifactId&gt;spring-kafka&lt;/artifactId&gt;
            &lt;version&gt;2.9.0&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
            &lt;groupId&gt;org.apache.kafka&lt;/groupId&gt;
            &lt;artifactId&gt;kafka-clients&lt;/artifactId&gt;
            &lt;version&gt;3.2.0&lt;/version&gt;
      &lt;/dependency&gt;
</code></pre>
<p>在application.yml配置文件中配置kafka</p>
<pre><code class="language-yaml">server:
port: 8080

spring:
kafka:
    bootstrap-servers: 云服务器外网ip地址:9092
    producer: # 生产者
      retries: 3 # 设置大于0的值，则客户端会将发送失败的记录重新发送
      batch-size: 16384
      buffer-memory: 33554432
      acks: 1
      # 指定消息key和消息体的编解码方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: default-group
      enable-auto-commit: false
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    listener:
      # 当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
      # RECORD
      # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交
      # BATCH
      # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交
      # TIME
      # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交
      # COUNT
      # TIME |　COUNT　有一个条件满足时提交
      # COUNT_TIME
      # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交
      # MANUAL
      # 手动调用Acknowledgment.acknowledge()后立即提交，一般使用这种
      # MANUAL_IMMEDIATE
      ack-mode: manual_immediate

</code></pre>
<p>生产者</p>
<pre><code class="language-java">@RestController
public class KafkaController {
    private final static String TOPIC_NAME = "test-topic";

    @Autowired
    private KafkaTemplate&lt;String, String&gt; kafkaTemplate;

    @RequestMapping("/send")
    public String send(@RequestParam("msg") String msg) {
      kafkaTemplate.send(TOPIC_NAME, "key", msg);
      return String.format("消息 %s 发送成功！", msg);
    }
}

</code></pre>
<p>消费者</p>
<pre><code class="language-java">@Component
public class DemoConsumer {
    /**
   * @param record record
   * @KafkaListener(groupId = "testGroup", topicPartitions = {
   * @TopicPartition(topic = "topic1", partitions = {"0", "1"}),
   * @TopicPartition(topic = "topic2", partitions = "0",
   * partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))
   * },concurrency = "6")
   * //concurrency就是同组下的消费者个数，就是并发消费数，必须小于等于分区总数
   */
    @KafkaListener(topics = "test-topic", groupId = "testGroup1")
    public void listentestGroup(ConsumerRecord&lt;String, String&gt; record, Acknowledgment ack) {
      String value = record.value();
      System.out.println("testGroup1 message: " + value);
      System.out.println("testGroup1 record: " + record);
      //手动提交offset，一般是提交一个banch，幂等性防止重复消息
      // === 每条消费完确认性能不好！
      ack.acknowledge();
    }

    //配置多个消费组
    @KafkaListener(topics = "test--topic", groupId = "testGroup2")
    public void listentestGroup2(ConsumerRecord&lt;String, String&gt; record, Acknowledgment ack) {
      String value = record.value();
      System.out.println("testGroup2 message: " + value);
      System.out.println("testGroup2 record: " + record);
      //手动提交offset
      ack.acknowledge();
    }
}

</code></pre>
<p>使用swagger测试发送消息</p>
<p><img src="https://img2022.cnblogs.com/blog/2286325/202211/2286325-20221114231208608-1843306412.png"></p>
<p>控制台打印消息</p>
<p><img src="https://img2022.cnblogs.com/blog/2286325/202211/2286325-20221114231216985-1463046945.png"></p><br><br>
来源：https://www.cnblogs.com/ndchao/p/chaos1.html
頁: [1]
查看完整版本: 云服务器(Linux)安装部署Kafka