（1）用户登录系统，在左侧列表页点击“大数据服务”——“Kafka消息队列”进入Kafka集群列表界面。

点击页面右上方的【新建集群】开始进行Kafka集群创建。

（2）进入Kafka集群创建界面后用户可以根据需求选取配置如图所示：

虚拟数据中心：选择可创建Kafka集群的虚拟数据中心，可以根据区域进行筛选选择。

计费方式：选择Kafka集群的计费方式：包年包月、按需计费。

版本：选择Kafka服务的版本：kafka-v0.10.2.1、kafka-v2.2.0。

私网网络：选择Kafka服务节点的通讯私网网络，默认占用所选VDC的私有网络IP地址。

节点机型：选择Kafka集群下节点的机型及计算配置。可选的机型有“高性能型”、“高IO型”。

数据盘：选择每个节点对应的数据盘。可选数据盘类型有性能型、超高性能型（SSD）云硬盘，超高性能型云硬盘（SSD）还可通过购买预置性能包提高IOPS及吞吐量。

节点数量：选择创建Kafka集群时要购买的kafka节点数量，节点数量最低为3台（默认前3台用于master、调度、备份，不支持删除此3台默认节点）

集群名称：设置Kafka集群的名称，最多支持输入36个字符。

消息保留时长：设置Kafka集群的消息保留的时间，默认保留24个小时，支持保留1-2160个小时。

配置及计费信息：当用户创建Kafka集群时可在创建界面右侧查看已选配置及计费，如下图所示

点击配置及计费窗口底部的【创建】按钮创建Kafka集群。

（1）在“集群详情”页面-基本信息模块中点击“集群名称”后的【编辑】按钮。

（2）在出现的编辑框中输入最新集群名称，可在“变更配置及计费信息”框中确认变更内容，点击【保存】按钮，完成更改。

（1）点击“运行状态”模块中的【调整自动续约】按钮，选择续约配置：

a. 关闭：关闭自动续约。Kafka集群到期以后不自动续费，集群到期后将被回收；

b. 开启：开启自动续约。Kafka集群到期后系统自动扣费进行续约；

1）开启-包年包月：继续按照包年包月进行续约，可选择每次续约的周期（续至月底/自然月）

2）开启-按需：当Kafka集群到期后将转为按需计费

（2）在右侧“变更配置及计费信息”模块中确认变更内容无误，点击【保存】按钮，完成变更。

点击“配置信息”模块右上角的【变更配置】按钮。调整消息保留时间，在“变更配置及计费信息”框中确认变更内容无误，点击【保存】按钮，完成变更。

（1）在节点管理页面，点击左上方【添加节点】按钮。

（2）在添加节点弹窗中，选择要添加的节点数量，确认价格，点击【确定】进行节点创建。

说明：添加的节点配置与创建Kafka集群时所选的配置一致，暂不支持指定单节点配置进行创建。

说明：

1. 默认Kafka集群有3个节点用于master、调度、备份，不支持删除；

2. 包年包月计费的kafka集群节点不支持删除；

点击页面右上方【新建Topic】按钮，在弹出的窗口中填写和选择topic信息：

Topic名称：设置topic的名称。支持输入字母/数字/下划线/./-，不可超过36个字符。

分区数量：选择topic的分区数量，支持1-100个分区。

副本数量：选择副本数量，支持2个/3个副本。

点击【确定】按钮，创建Topic。

选择要删除Topic右侧操作列的【删除】按钮，在删除确认弹窗中点击【确定】按钮，删除topic。

（1）在Consumer Group页面点击消费组左侧的“展开”按钮，可以展开分区状态信息。

（2）在展开的分区状态内容，可查看对应topic的分区数量、提交的offset位置、最大的offset位置、未消费消息条数。

（1）选择对应消费组操作列的【offset设置】按钮

（2）在弹窗中选择topic、offset位置，点击【确定】，完成offset设置操作。

控制台是首云提供的Kafka Manager工具，用于辅助管理Kafka集群，可进行监控数据查询、高级管理操作。

点击要删除Kafka集群操作列的【删除】按钮，在弹出的删除确认弹窗中，点击【确定】按钮，完成删除集群操作。

注意：删除集群会清除其下所有资源，风险较大，建议您开启资源删除验证，在删除Kafka集群时，会进行身份验证，完成验证后才可成功执行集群删除任务。

用户中心——设置——附加设置，取消验证的操作设置为【禁用】状态，则删除资源时将进行身份验证才可删除。

在Kafka消息队列集群列表页面，点击页面右上方的【告警管理】按钮，进入告警管理页面。

（1）在“Kafka消息队列告警管理”页面，点击右上方【新建告警】按钮，进入告警创建页面。

（2）在新建告警页面，填写告警内容，如下图所示：

在“Kafka消息队列告警管理”页面，可对已创建的告警策略进行基本管理操作。

（1）选择某一告警策略，点击“操作”列的【复制】按钮，进入“新建告警”页面。

（2）在新建告警页面会发现，已经继承了对应的告警内容。在此基础上进行修改后，点击【新建】按钮，可快速创建一个新的告警。

点击“Kafka消息队列告警管理”右上方【查看历史告警记录】按钮，在出现的弹窗中，可按时间查询此客户的kafka告警历史。包含信息：告警对象（Kafka集群名称）、告警内容、发送时间。

在“Kafka消息队列告警管理”页面的列表中，选择某一告警策略，在“操作”列中，点击【删除】按钮。在弹出的“删除告警”确认页面中，点击【确认】按钮，完成删除。

SSL客户端适配

首云kafka支持通过SSL加密和认证，并且可以通过该方式使集群具有出公网的能力。 开通方式： 在首云控制台创建kafka集群时，选择公网实例，并且选择绑定客户公网ip，则默认打开SSL安全认证；在集群详情的数据安全栏可以看到公网连接的地址和加密认证文件下载按钮

kafka自带客户端

1，在页面下载SSL证书kafka.truststore和 kafka.keystore 放置到客户端某一目录下，此处为/usr/local/ssl_store/

[root@kafka-cds kafka]# ls /usr/local/ssl_store/
kafka.keystore  kafka.truststore

2，创建客户端配置文件client_ssl.properties

[root@kafka-cds kafka]# cat client_ssl.properties
# 这里是SSL连接地址，使用的是9093端口
bootstrap.servers=......kafka.yun-paas.com.:9093 
# 使用的是SSL协议
security.protocol=SSL
# truststore证书路径
ssl.truststore.location=/usr/local/ssl_store/kafka.truststore
# truststore密码：统一为kafkasslpwd
ssl.truststore.password=kafkasslpwd 
# keystore证书路径
ssl.keystore.location=/usr/local/ssl_store/kafka.keystore 
# keystore密码：统一为kafkasslpwd
ssl.keystore.password=kafkasslpwd
# 为了兼容kafka v2.0+
ssl.endpoint.identification.algorithm=   

3，生产者客户端命令，配置文件为client_ssl.properties：

[root@kafka-cds kafka]#  bin/kafka-console-producer.sh --broker-list ......kafka.yun-paas.com.:9093  --topic ssltest --producer.config ./client_ssl.properties

4，消费者客户端命令，配置文件为client_ssl.properties(与生产者的配置文件相同)：

[root@kafka-cds kafka]#  bin/kafka-console-consumer.sh --bootstrap-server ......kafka.yun-paas.com.:9093 --topic ssltest  --consumer.config ./client_ssl.properties

Java客戶端

1, 导入maven包

<dependency>  
  <groupId>org.apache.kafka</groupId>  
  <artifactId>kafka-clients</artifactId>  
  <version>0.10.0.1</version>
</dependency>

2, 导入依赖并创建Properties

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.config.SslConfigs;

Properties props = new Properties();
# 这里是SSL连接地址，使用的是9093端口
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "......kafka.yun-paas.com.:9093");

# 使用的是SSL协议
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
# truststore证书路径
props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "/usr/local/ssl_store/kafka.truststore");
 # truststore密码：统一为kafkasslpwd
props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,  "kafkasslpwd");
# keystore证书路径
props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, 
"/usr/local/ssl_store/kafka.keystore");
 # keystore密码：统一为kafkasslpwd
props.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, "kafkasslpwd");
# 为了兼容kafka v2.0+
props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_DOC, "");

python 客户端

参考： http://maximilianchrist.com/python/databases/2016/08/13/connect-to-apache-kafka-from-python-using-ssl.html

选用pykafka库，安装方法：

pip install pykafka

1，从kafka公网实例详情页面下载kafka.keystore 文件，上传到客户端服务器上。

2，提取密钥

• 2.1 配置完Apache Kafka实例后，您将拥有两个JKS容器：“ kafka.keystore”和“ kafka.truststore”。第一个包含已签名的客户端证书，其私钥和用于对其进行签名的“ CARoot”证书。第二个证书包含用于签署客户端证书和密钥的证书。因此，我们需要的所有内容都包含在“ kafka.client.keystore.jks”文件中。要获得其内容的概述 , 可以查看：

keytool -list -rfc -keystore kafka.keystore

• 2.2 现在开始工作。首先，我们将提取客户证书：

keytool -exportcert -alias cds-cluster  -keystore \
kafka.keystore -rfc -file certificate.pem

• 2.3 接下来，我们将提取客户密钥。并不直接支持此功能keytool，这就是为什么我们必须首先将密钥库转换为pkcs12格式，然后从中提取私钥： 第二条命令仅将密钥输出到STDOUT。从那里可以将其复制并粘贴到“ key.pem”中。确保复制在“ -BEGIN PRIVATE KEY”-和“ -END PRIVATE KEY”-之间的所有行。

keytool -v -importkeystore -srckeystore kafka.keystore \
-srcalias cds-cluster -destkeystore cert_and_key.p12 -deststoretype PKCS12

openssl pkcs12 -in cert_and_key.p12 -nocerts -nodes

• 2.4 最后，我们将提取CARoot证书：

keytool -exportcert -alias CARoot -keystore kafka.client.keystore.jks -rfc \
          -file CARoot.pem

此时，证书文件kafka.keystore所在路径下会有3个文件：CARoot.pem，key.pem， certificate.pem

3，配置python客户端：

from pykafka import KafkaClient, SslConfig
import logginglogging.basicConfig(level=logging.DEBUG)

config = SslConfig(
            # 分别配置3个密钥文件的路径
          cafile="/root/CARoot.pem",
          certfile='/root/certificate.pem',
          keyfile='/root/key.pem',
            # 密码默认都是kafkasslpwd
         password=b"kafkasslpwd"
            )
# 集群详情页面的SSL连接地址
hosts = "......kafka.yun-paas.com.:9093,......yun-paas.com.:9093"

# 配置client, broker_version 试版本而定
client=KafkaClient(hosts=hosts,
                   ssl_config=config,
                   broker_version='2.2.0')
                   
topic = client.topics["ssltest"]
# Write hello world to ssltest topicwith 

topic.get_sync_producer() as producer:
    producer.produce(b'Hello pykafka ssl')
# Print all messages from ssltest 

topicconsumer = topic.get_simple_consumer()

for message in consumer:
    if message is not None:
        print('{} {}'.format(message.offset, message.value))

备注： consumer 消费时需要安装python-snappy;

pip3 install python-snappy

会出现很多依赖问题:

1，找不到snappy.h
下载源码 snappy-1.1.3.tar.gz
解压，编译
tar -zxvf snappy-1.1.3.tar.gz
cd snappy-1.1.3
./configure
make && make install

2，找不到Python.h
yum search python | grep -i devel
yum install python36-devel.x86_64 

Flume 接入kafka公网(SSL加密)

请选择apache-flume-1.8.0-bin 版本 1,下载apache-flume-1.8.0-bin.tar.gz文件，解压为 apache-flume-1.8.0-bin 2，进入apache-flume-1.8.0-bin目录，在conf目录下新建kafka.conf 文件

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /tmp/logs
a1.sources.r1.fileHeader = true

# Describe the sink
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = ssltest
# 页面SSL连接地址
a1.sinks.k1.kafka.bootstrap.servers =  ......-11ea.kafka.yun-paas.com.:9093...
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 1
a1.sinks.k1.kafka.producer.compression.type = snappy

a1.sinks.k1.kafka.producer.security.protocol = SSL
# kafka.truststore证书的路径
a1.sinks.k1.kafka.producer.ssl.truststore.location = /tmp/ssl/kafka.truststore
# kafka.truststore证书的密码，统一为kafkasslpwd
a1.sinks.k1.kafka.producer.ssl.truststore.password = kafkasslpwd
# kafka.keystore证书的路径
a1.sinks.k1.kafka.producer.ssl.keystore.location = /tmp/ssl/kafka.keystore
# kafka.keystore证书的密码，统一为kafkasslpwd
a1.sinks.k1.kafka.producer.ssl.keystore.password = kafkasslpwd
a1.sinks.k1.kafka.producer.ssl.endpoint.identification.algorithm=
a1.sinks.k1.kafka.producer.ssl.key.password=kafkasslpwd

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1