Kafka尝试多次以确保大文件下载成功

欢迎各位小伙伴来到golang学习网，相聚于此都是缘哈哈哈！今天我给大家带来《Kafka尝试多次以确保大文件下载成功》，这篇文章主要讲到等等知识，如果你对Golang相关的知识非常感兴趣或者正在自学，都可以关注我，我会持续更新相关文章！当然，有什么建议也欢迎在评论留言提出！一起学习！

我是 kafka 的新手，我尝试构建一个服务，发送带有附加文件的邮件。 执行流程：

• kafka会收到消息发送邮件
• 函数 get file 将从 url 下载文件，缩放图像并保存文件
• 发送邮件时，我将从文件夹中获取文件并附加到表单 问题：
• 当我多次发送大文件邮件时，kafka重试多次，我会收到很多邮件

kafka错误：“kafka服务器：提供的成员在当前一代中未知”

我听了 maxprocessingtime ，但我尝试测试带有大文件的邮件，它仍然工作正常

kafka 信息：1 个经纪人，3 个消费者

func (s *customermailservice) sendpodmail() error { filepaths, err := downloadfiles(podurls, orderinfo.ordercode)

if err != nil{
    countretry := 0
    for countretry <= num_of_retry{
        filepaths, err = downloadfiles(podurls, orderinfo.ordercode)
        if err == nil{
            break
        }
        countretry++
    }
}

    err = s.sendmailservice.send(ctx, orderinfo.customeremail, tmspod, content,filepaths)}

函数下载文件：

func downloadfiles(files []string, ordercode string) ([]string, error) {
var filepaths []string

err := os.mkdir(tempdir, 0750)
if err != nil && !os.isexist(err) {
    return nil, err
}

tempdirpath := tempdir + "/" + ordercode
err = os.mkdir(tempdirpath, 0750)
if err != nil && !os.isexist(err) {
    return nil, err
}

for _, fileurl := range files {
    fileurlparsed, err := url.parserequesturi(fileurl)
    if err != nil {
        logrus.witherror(err).infof("pod url is invalid %s", ordercode)
        return nil, err
    }

    extfile := filepath.ext(fileurlparsed.path)
    dir, err := os.mkdirtemp(tempdirpath, "tempdir")

    if err != nil {
        return nil, err
    }

    f, err := os.createtemp(dir, "tmpfile-*"+extfile)
    if err != nil {
        return nil, err
    }
    defer f.close()

    response, err := http.get(fileurl)
    if err != nil {
        return nil, err
    }
    defer response.body.close()

    contenttypes := response.header["content-type"]
    istypeallow := false
    for _, contenttype := range contenttypes {
        if contenttype == "image/png" || contenttype == "image/jpeg" {
            istypeallow = true
        }
    }

    if !istypeallow {
        logrus.witherror(err).infof("pod image type is invalid %s", ordercode)
        return nil, errors.new("pod image type is invalid")
    }

    decodedimg, err := imaging.decode(response.body)
    if err != nil {
        return nil, err
    }

    resizedimg := imaging.resize(decodedimg, 1024, 0, imaging.lanczos)

    imaging.save(resizedimg, f.name())

    filepaths = append(filepaths, f.name())
}
return filepaths, nil}

发送邮件功能

func (s *tikiMailService) SendFile(ctx context.Context, receiver string, templateCode string, data interface{}, filePaths []string) error {
path := "/v1/emails"
fullPath := fmt.Sprintf("%s%s", s.host, path)

formValue := &bytes.Buffer{}
writer := multipart.NewWriter(formValue)
_ = writer.WriteField("template", templateCode)
_ = writer.WriteField("to", receiver)

if data != nil {
    b, err := json.Marshal(data)
    if err != nil {
        return errors.Wrapf(err, "Cannot marshal mail data to json with object %+v", data)
    }

    _ = writer.WriteField("params", string(b))
}

for _, filePath := range filePaths {
    part, err := writer.CreateFormFile(filePath, filepath.Base(filePath))

    if err != nil {
        return err
    }


    pipeReader, pipeWriter := io.Pipe()

    go func() {
        defer pipeWriter.Close()

        file, err := os.Open(filePath)
        if err != nil {
            return 
        }
        defer file.Close()

        io.Copy(pipeWriter, file)
    }()

    io.Copy(part, pipeReader)
}

err := writer.Close()
if err != nil {
    return err
}

request, err := http.NewRequest("POST", fullPath, formValue)
if err != nil {
    return err
}
request.Header.Set("Content-Type", writer.FormDataContentType())

resp, err := s.doer.Do(request)
if err != nil {
    return errors.Wrap(err, "Cannot send request to send email")
}
defer resp.Body.Close()

b, err := ioutil.ReadAll(resp.Body)
if err != nil {
    return err
}

if resp.StatusCode != http.StatusOK {
    return errors.New(fmt.Sprintf("Send email with code %s error: status code %d, response %s",
        templateCode, resp.StatusCode, string(b)))
} else {
    logrus.Infof("Send email with attachment ,code %s success with response %s , box-code", templateCode, string(b),filePaths)
}
return nil
}

谢谢

正确答案

我的团队在重新部署 k8s pod 时发现了问题，这导致领导分区冲突，导致重新平衡。它会再次尝试处理 pod 缓冲区中剩余的消息。

解决方案：我没有获取缓冲区中保存的许多消息，我只是获取一条消息并通过配置处理它：

channelbuffersize = 0

冲突领导者分区示例：

consumer A and B startup in the same time
consumer A registers itself as leader, and owns the topic with all partitions
consumer B registers itself as leader, and then begins to rebalance and owns all partitions
consumer A rebalance and obtains all partitions, but can not consume because the memberId is old and need a new one
consumer B rebalance again and owns the topic with all partitions, but it's already obtained by consumer A

我的两分钱：如果附件非常大，消费者需要花费大量时间来读取文件并将其作为附件发送。

这会增加两次 poll() 调用之间的时间量。如果该时间大于 max.poll.interval.ms，则认为使用者失败并且不会提交分区偏移量。结果，消息被再次处理，最终，如果执行时间偶然低于轮询间隔，则提交偏移量。效果是发送多封电子邮件。

尝试增加消费者端的 max.poll.interval.ms。

理论要掌握，实操不能落！以上关于《Kafka尝试多次以确保大文件下载成功》的详细介绍，大家都掌握了吧！如果想要继续提升自己的能力，那么就来关注golang学习网公众号吧！