package main

import (
    "crypto/tls"
    "crypto/x509"
    "fmt"
    "github.com/Shopify/sarama"
    "io/ioutil"
    "log"
    "time"
)

//var AddressSSL1 = []string{"172.18.236.78:9092", "172.18.236.79:9092", "172.18.236.80:9092"}
var AddressSSL1 = []string{"172.18.14.215:9092", "172.18.14.215:9093"}

func main() {

    for i := 0; i < 10; i++ {
        KafkaProducer(AddressSSL1, "testSSl", "a")

    }
}

func KafkaProducer(address []string, topic string, sendValue string) {
    fmt.Println("producer 启动")
    //设置配置
    config := sarama.NewConfig()
    //加载证书
    tlsConfig, err := NewTLSConfigProducer("./ssl/client.cer.pem",
        "./ssl/client.key.pem", "./ssl/server.cer.pem")
    if err != nil {
        log.Fatal(err)
    }
    //客户端不对服务端验证
    tlsConfig.InsecureSkipVerify = true
    //启动SSL认证
    config.Net.TLS.Enable = true
    config.Net.TLS.Config = tlsConfig
    //是否等待成功和失败后的响应,只有上面的RequireAcks(等待服务器所有副本都保存成功后的响应)设置不是NoReponse才有用
    config.Producer.Return.Successes = true
    config.Producer.Timeout = 5 * time.Second
    client, err := sarama.NewClient(address, config)
    if err != nil {
        log.Fatalf("unable to create kafka client: %q", err)
    }
    //使用配置,新建一个异步生产者
    p, err := sarama.NewAsyncProducerFromClient(client)
    if err != nil {
        log.Fatal(err)
    }
    defer p.Close()
    loopProducer(p, topic, sendValue)
}
func NewTLSConfigProducer(clientCertFile, clientKeyFile, caCertFile string) (*tls.Config, error) {
    tlsConfig := tls.Config{}

    // 加载客户端证书
    cert, err := tls.LoadX509KeyPair(clientCertFile, clientKeyFile)
    if err != nil {
        return &tlsConfig, err
    }
    tlsConfig.Certificates = []tls.Certificate{cert}

    // 加载CA证书
    caCert, err := ioutil.ReadFile(caCertFile)
    if err != nil {
        return &tlsConfig, err
    }
    caCertPool := x509.NewCertPool()
    caCertPool.AppendCertsFromPEM(caCert)
    tlsConfig.RootCAs = caCertPool

    tlsConfig.BuildNameToCertificate()
    return &tlsConfig, err
}

func loopProducer(producer sarama.AsyncProducer, topic string, text string) {

    text = string(time.Now().Format("15:04:05"))

    msg := &sarama.ProducerMessage{
        Topic:     topic,
        Key:       nil,
        Value:     sarama.StringEncoder(text),
        Headers:   nil,
        Metadata:  nil,
        Offset:    0,
        Partition: 0,
        Timestamp: time.Now(),
    }

    producer.Input() <- msg
    log.Printf("Produced message: [%s] - %s\n", text,msg.Timestamp.Format("15:04:05"))
}

package main

import (
    "crypto/tls"
    "crypto/x509"
    "github.com/Shopify/sarama"
    "io/ioutil"
    "log"
    "os"
    "os/signal"
    "sync"
)

//var AddressSSL2 = []string{"172.18.236.78:9092", "172.18.236.79:9092", "172.18.236.80:9092"}
var AddressSSL2 = []string{"172.18.14.215:9092","172.18.14.215:9093"}

func main() {
    KafkaConsumer(AddressSSL2, "testSSl")
}

func KafkaConsumer(address []string, topics string) {
    tlsConfig, err := NewTLSConfigConsumer("./ssl/client.cer.pem",
        "./ssl/client.key.pem", "./ssl/server.cer.pem")
    if err != nil {
        log.Fatal(err)
    }

    //客户端不对服务端验证
    tlsConfig.InsecureSkipVerify = true

    //创建一个配置对象
    consumerConfig := sarama.NewConfig()
    //启动TLS通讯
    consumerConfig.Net.TLS.Enable = true
    consumerConfig.Net.TLS.Config = tlsConfig
    consumerConfig.Version = sarama.V2_3_0_0

    //创建一个客户消费者
    client, err := sarama.NewClient(address, consumerConfig)
    if err != nil {
        log.Fatalf("unable to create kafka client: %q", err)
    }

    consumer, err := sarama.NewConsumerFromClient(client)
    if err != nil {
        log.Fatal(err)
    }
    defer consumer.Close()

    consumerLoop(consumer, topics)
}

func consumerLoop(consumer sarama.Consumer, topic string) {
    partitions, err := consumer.Partitions(topic)
    if err != nil {
        log.Println("unable to fetch partition IDs for the topic", topic, err)
        return
    }

    signals := make(chan os.Signal, 1)
    signal.Notify(signals, os.Interrupt)

    var wg sync.WaitGroup
    for partition := range partitions {
        wg.Add(1)
        go func() {
            consumePartition(consumer, int32(partition), signals, topic)
            wg.Done()
        }()
    }
    wg.Wait()
}

func NewTLSConfigConsumer(clientCertFile, clientKeyFile, caCertFile string) (*tls.Config, error) {
    tlsConfig := tls.Config{}

    // 加载客户端证书
    cert, err := tls.LoadX509KeyPair(clientCertFile, clientKeyFile)
    if err != nil {
        return &tlsConfig, err
    }
    tlsConfig.Certificates = []tls.Certificate{cert}

    // 加载CA证书
    caCert, err := ioutil.ReadFile(caCertFile)
    if err != nil {
        return &tlsConfig, err
    }
    caCertPool := x509.NewCertPool()
    caCertPool.AppendCertsFromPEM(caCert)
    tlsConfig.RootCAs = caCertPool

    tlsConfig.BuildNameToCertificate()
    return &tlsConfig, err
}

func consumePartition(consumer sarama.Consumer, partition int32, signals chan os.Signal, topic string) {
    log.Println("Receving on partition", partition)
    partitionConsumer, err := consumer.ConsumePartition(topic, partition, sarama.OffsetOldest)
    if err != nil {
        log.Println(err)
        return
    }
    defer func() {
        if err := partitionConsumer.Close(); err != nil {
            log.Println(err)
        }
    }()

    consumed := 0
ConsumerLoop:
    for {
        select {
        case msg := <-partitionConsumer.Messages():
            //log.Printf("Consumed message offset %d\nData: %s\n", msg.Offset, msg.Value)
            t := msg.Timestamp
            formatTime := t.Format("15:04:05")
            log.Printf("%s - %s\n", msg.Value,formatTime)
            consumed++
        case <-signals:
            break ConsumerLoop
        }
    }
    log.Printf("Consumed: %d\n", consumed)
}