云服务器(Linux)安装部署Kafka的详细过程

目录
  • 云服务器(Linux)安装部署Kafka
    • 前期准备
    • 下载安装包
    • 上载安装包到云服务器
    • 配置kafka
    • 开放云服务器端口
    • 开放linux防火墙端口
    • 启动kafka服务
    • 测试单机连通性
    • Springboot连接kafak

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

前期准备

kafka的安装需要依赖于jdk,需要在服务器上提前安装好该环境,这里使用用jdk1.8。

下载安装包

官网地址:

较新的版本已自带Zookeeper,无需额外下载。这里使用3.2.0做演示。

注意要下载Binary downloads标签下的tgz包,Source download标签下的包为源码。无法直接运行,需要编译。

上载安装包到云服务器

使用ssh连接工具将kafka_2.12-3.2.0.tgz这个包上传到云服务器上的一个目录。

打开命令行,进入到放有压缩包的目录,执行

tar -zxvf kafka_2.12-3.2.0.tgz

配置kafka

然后使用cd命令进入到/kafka_2.12-3.2.0/config/下,使用

vi server.properties

编辑配置文件。

删除listeners和advertised前方的#号,改成如下配置:

listeners=PLAINTEXT://云服务器内网ip:9092(本地访问用本地ip)
# 如果要提供外网访问则必须配置此项
advertised.listeners=PLAINTEXT://云服务器公网ip:9092(若要远程访问需配置此项为云服务器的公网ip)
# zookeeper连接地址,集群配置格式为ip:port,ip:port,ip:port
zookeeper.connect=云服务器公网ip:2181

开放云服务器端口

在云服务器控制台内进入安全组页面,添加两条新的入站规则,tcp/9092和tcp/2181

开放linux防火墙端口

先查看使用的防火墙类型iptables/firewalld

iptables操作命令

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

Centos7默认安装了firewalld,如果没有安装的话,可以使用 yum install firewalld firewalld-config进行安装。

操作指令如下:

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)

打开tcp/9092和tcp/2181这两个端口后,重启防火墙,并查看开放的端口确实生效。

启动kafka服务

cd命令进入kafka_2.12-3.2.0目录下,执行

bin/zookeeper-server-start.sh config/zookeeper.properties

启动zookeeper,不加-daemon方便排除启动错误,新建一个shell窗口,进入该目录再执行

bin/kafka-server-start.sh config/server.properties

启动kafka,若打印日志未报错,若未出现error日志,说明启动成功。

测试单机连通性

查询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

Springboot连接kafak

在pom.xml文件中引入kafka依赖

<dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.9.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>3.2.0</version>
        </dependency>

在application.yml配置文件中配置kafka

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

生产者

@RestController
public class KafkaController {
    private final static String TOPIC_NAME = "test-topic";

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

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

消费者

@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<String, String> 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<String, String> record, Acknowledgment ack) {
        String value = record.value();
        System.out.println("testGroup2 message: " + value);
        System.out.println("testGroup2 record: " + record);
        //手动提交offset
        ack.acknowledge();
    }
}

使用swagger测试发送消息

控制台打印消息

到此这篇关于云服务器(Linux)安装部署Kafka的详细过程的文章就介绍到这了,更多相关Linux安装Kafka内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们!

时间: 2022-11-12

阿里云云服务器Linux系统FTP服务器搭建设置教程

一.Redhat/CentOS安装vsftp软件 1.更新yum源 首先需要更新系统的yum源,便捷工具下载地址:http://www.jb51.net/article/54840.htm(shell脚本) 2.安装vsftp 使用yum命令安装vsftp 复制代码 代码如下: #yum install vsftpd -y 3.添加ftp帐号和目录 先检查一下nologin的位置,通常在/usr/sbin/nologin或者/sbin/nologin下. 使用下面的命令创建帐户,该命令指定了/a

阿里云云服务器Linux系统更新yum源Shell脚本

所有执行的脚本都需要root身份来执行,执行方法:以root身份执行命令:bash xxx.sh 功能:自动检测系统并更新源 适用系统版本:兼容线上所有linux版本 执行方法:以root身份执行命令:bash update_source.sh 解决了什么问题:一键式检测系统并更新源 给客户带来了好处:用户只需执行该脚本一次即可自动检测系统并更新源. 备注:由于系统版本都有支持的周期所以部分源可能会出现不可用的情况,包括官方的源,这是正常情况 #!/bin/bash ##############

Linux下Kafka单机安装配置方法(图文)

介绍 Kafka是一个分布式的.可分区的.可复制的消息系统.它提供了普通消息系统的功能,但具有自己独特的设计.这个独特的设计是什么样的呢? 首先让我们看几个基本的消息系统术语: •Kafka将消息以topic为单位进行归纳. •将向Kafka topic发布消息的程序成为producers. •将预订topics并消费消息的程序成为consumer. •Kafka以集群的方式运行,可以由一个或多个服务组成,每个服务叫做一个broker. producers通过网络将消息发送到Kafka集群,集群

在Linux环境下安装Kafka

目录 二.生产与消费 2.1 kafka-topics.sh 用于管理主题 2.2 kafka-console-consumer.sh用于消费消息 2.3 kafka-console-producer.sh用于生产消息 2.4 具体操作 一.环境准备 jdk下载地址链接:下载地址 zookeeper下载地址链接:下载地址 kafka下载地址链接:下载地址 1.1 Java环境为前提 1.1.1 上传jdk-8u261-linux-x64.rpm到服务器并安装 # 安装命令 rpm -ivh jd

阿里云ECS云服务器(linux系统)安装mysql后远程连接不了(踩坑)

昨天买了一年的阿里云服务器,系统是linux Centos7的,满怀憧憬的装了个mysql,接下来的一天让我差点怀疑人生... 怎么装mysql就不多说了,反正我装了三遍,每次在阿里云上都能本地连接数据库 用navcat远程连阿里云的数据库死活连接不上.始终报:2003 - Can't connect to MySQL SERVER ON ********* (10060) 装了三次你就知道我有多绝望了,因为第一次linux下安装mysql,每次都认为可能是安装出了问题, 百度都被我翻烂了...

Linux一键安装web环境全攻略(阿里云服务器)

一键安装包下载: 点此下载 安装须知 1.此安装包可在阿里云所有linux系统上部署安装,此安装包包含的软件及版本为: nginx:1.0.15.1.2.5.1.4.4 apache:2.2.22.2.4.2 mysql:5.1.73.5.5.35.5.6.15 php:5.3.18.5.4.23.5.5.7 php扩展:memcache.Zend Engine/ OPcache ftp:(yum/apt-get安装) phpwind:8.7 GBK phpmyadmin:4.1.8 2.请使用

Linux使用scp远程传输命令进行服务器文件互传(阿里云服务器)

前几天 WordPress大学 所在的服务器环境不太稳定,CPU 占用总是 100%,折腾了好几次都没有找到原因,只好考虑重新配置环境,考虑到配置和优化好环境需要几个小时,总不能将网站停掉吧,所以要将 WordPress大学 网站临时迁移到另外的服务器运行,配置好再迁回来. 看来下阿里云的两个云服务器,都是同一个区的,所以决定通过内网进行文件传输,速度快不说(正常传输文件的速度能达到35m~50m/s),还不占用公网带宽. 阿里云云服务器有一个安全组策略,可能需要设置下 内网入.出方向的IP白名

Linux本机与服务器文件互传及Linux服务器文件上传下载命令写法

一.Linux下文件互传,scp命令实例 1.Linux下目录复制:本机->远程服务器 scp -r /home/abc/test1 root@192.168.0.1:/home/bcd/test2 (本机目录路径 远程机用户名@IP:/目录) 2.Linux下目录复制:远程服务器->本机 scp -r root@192.168.0.1:/home/bcd/test2 /home/abc/test1 (远程机用户名@IP:/目录 本机目录路径) 3.Linux下文件复制:远程服务器->本

详细部署阿里云服务器全过程(图文教程)

最近学习了Web开发前后端等技术,便想着将项目部署到云服务器,方便后续管理,顺便学习一下部署云服务器的过程. 购买与连接云服务器 部署云服务器,首先需要的便是购买云服务器,这里我选择的是阿里云服务器,注册实名认证这里就不详细说明了,我购买的是云服务器ECS,可以选择大学生优惠,一个月9.5元,算是很实惠的. 我选择的操作系统是Linux版的 centos_7_03_64,后续还可以自己去购买一个域名. 购买完之后可以点击图里的更多进行重置密码,我这里已经重置过了,就不给大家演示了. 接下来就是管

linux下scp和sftp命令使用详解

目录 前言 1. scp 使用 2. sftp 使用 3. scp 和 sftp 有啥区别?用哪个好? 总结 前言 scp 和 sftp 是一种远程文件加密传输协议,讲通俗点就是用来操控本地/远程文件,不知道你注意到了没有,它们开头都带了 s,是的,它们嵌套了一层 ssh 加密协议. 像 sftp 它的传统版本就是 ftp,但这种协议不安全,传输数据是都是明文的,很容易受到攻击窃取,所以才有了后来的 sftp,其实就跟 http / https 同个道理. 如果你还没了解过 ssh 可以参考我之

Linux阿里云服务器中安装Nginx命令的详细过程

目录 1. 安装nginx依赖 2. wget下载nginx 3. 解压缩 4. 自动配置 5. 执行make命令 6. 配置环境变量 7. 查看nginx错误日志 1. 安装nginx依赖 yum -y install gcc zlib zlib-devel pcre-devel openssl openssl-devel 2. wget下载nginx 在新建的nginx文件夹中下载nginx,版本可以在Nginx官网自选我选择了当前时间点官网中最新最稳定的版本 wget http://ngi

使用paramiko远程执行命令、下发文件的实例

写部署脚本时,难免涉及到一些远程执行命令或者传输文件. 之前一直使用sh库,调用sh.ssh远程执行一些命令,sh.scp传输文件,但是实际使用中还是比较麻烦的,光是模拟用户登陆这一点,还需要单独定义方法模拟输入. 感受一下: from sh import ssh PASS = 'xxxx' def ssh_interact(line, stdin): line = line.strip() print(line) if line.endswith('password:'): stdin.put

阿里云服务器linux系统搭建Tomcat部署Web项目

整个过程我给它分成四个步骤: 下载并安装jdk 下载并安装Tomcat 配置阿里云服务器信息 部署web项目 使用的工具:Xshell.WinSCP. 没有安装jdk的小伙伴点击上方超链接跳转到安装jdk博客 下载并安装Tomcat 到这个网页下查看最新的镜像:https://mirrors.tuna.tsinghua.edu.cn/apache/tomcat 使用工具Xshell操作Linux系统 移动到home目录下载tomcat 下载 wget https://mirrors.tuna.t

在 Linux 中不使用 CD 命令进入目录/文件夹的方法

众所周知,如果没有 cd 命令,我们无法 Linux 中切换目录.这个没错,但我们有一个名为 shopt 的 Linux 内置命令能帮助我们解决这个问题. shopt 是一个 shell 内置命令,用于设置和取消设置各种 bash shell 选项,由于它已安装,因此我们不需要再次安装它. 是的,我们可以在启用此选项后,可以不使用 cd 命令切换目录. 我们将在本文中向你展示如何操作.这是一个小的调整,但对于那些从 Windows 迁移到 Linux 的新手来说非常有用. 这对 Linux 管理

javaWeb项目部署到阿里云服务器步骤详解

记录web项目部署到阿里云服务器步骤 (使用 web项目.阿里云服务器.Xftp.Xshell),敬请参考和指正 1.将要部署的项目打包成WAR文件格式,可以在MyEclipse.Eclipse都可以完成打包,如下图: 2.安装Xshell和Xftp两种软件 简单介绍下这两种软件作用(详情请百度相关文档) Xshell:通过网络连接到远程服务器主机. Xftp:能在Linux.Unix和Windows之间互传文件. 3.通过Xshell连接远程主机,如下图    4.创建会话完成,点击连接,显示

SpringBoot项目部署到阿里云服务器的实现步骤

目录 一.申请阿里云服务器 二.Xshell 的安装使用 三.云服务器上jdk的安装 四.linux云服务器安装mysql 五.部署SpringBoot项目 SpringBoot项目部署到阿里云linux服务器全流程 文章里所有需要的软件.jdk.mysql.xshell等下载地址: Xshell 5 下载地址:https://www.jb51.net/softs/56322.html jdk下载地址:https://www.jb51.net/softs/698365.html tomcat下载