阿里云函数计算对接kafka实战

vlambda
2021-07-29

阿里云函数计算对接kafka实战

新钛云服已为您服务 1068 天

背景需求

阿里云函数计算式是按调用次数来计算费用的，无需服务器就能进行后端的一些处理，对于调用次数不是特别多的场景比较适用。可以节省成本，但是如果调用次数很多对服务器性能要求不是特别高的情况下建议还是用ECS来部署服务。

函数计算对接kafka实战

新建服务和函数

新建服务比较容易基本只要输入一个服务名称即可

新建函数使用HTTP 函数，新手也可以尝试使用模版函数

配置函数

这里主要注意函数入口这个配置，Handler 的格式为 [文件名].[函数名]。例如创建函数时指定的 Handler 为 index.handler，那么文件名为 index.py，入口函数为 handler。

配置触发器

这里只是为了测试所以认证方式使用anonymous。

代码开发

首先要在代码根目录下安装python连接kafka的依赖包

pip install -t . kafka-python

然后开始编写setting.py这个是连接kafka的配置信息

vim setting.py

 
   
   
 
  
    
    
  kafka_setting = {
   'bootstrap_servers': ["XXX", "XXX", "XXX"],   #kafka连接地址
   'topic_name': 'XXX', #使用的topic名称
   'consumer_id': 'XXX' #使用的Consumer Group
}

然后开始编写入口文件index.py

 
   
   
 
  
    
    
  # -*- coding: utf-8 -*-
# 导入连接kafka所需依赖包和配置
import socket
from kafka import KafkaProducer
from kafka.errors import KafkaError

import setting

conf = setting.kafka_setting

print conf

HELLO_WORLD = b"Hello world!\n"

def handler(environ, start_response):    
   context = environ['fc.context']    
   request_uri = environ['fc.request_uri']    
   for k, v in environ.items():        
       if k.startswith("HTTP_"):            
           # process custom request headers            
           pass    

   # get request_body    
   try:        
       request_body_size = int(environ.get('CONTENT_LENGTH', 0))    
   except (ValueError):        
       request_body_size = 0   
   request_body = environ['wsgi.input'].read(request_body_size)   

   # get request_method    
   request_method = environ['REQUEST_METHOD']    

   # get path info    
   path_info = environ['PATH_INFO']    

   # get server_protocol    
   server_protocol = environ['SERVER_PROTOCOL']    

   # get content_type    
   try:        
       content_type = environ['CONTENT_TYPE']    
   except (KeyError):        
       content_type = " "    

   # get query_string    
   try:        
       query_string = environ['QUERY_STRING']        
   except (KeyError):        
       query_string = " "   

   print 'request_body: {}'.format(request_body)   
   print 'method: {}\n path: {}\n query_string: {}\n server_protocol: {}\n'.format(request_method, path_info, query_string, server_protocol)    
   # do something here    

   status = '200 OK'    
   response_headers = [('Content-type', 'text/plain')]    
   start_response(status, response_headers)

   #以下是kafka操作部分，发送一个消息到kafka
   producer = KafkaProducer(bootstrap_servers=conf['bootstrap_servers'],
                       api_version = (0,10),
                       retries=5)

   partitions = producer.partitions_for(conf['topic_name'])
   print 'Topic 下分区: %s' % partitions

   try:
       future = producer.send(conf['topic_name'], 'hello aliyun-kafka test!')
       future.get()
       print 'send message succeed.'
   except KafkaError, e:
       print 'send message failed.'
       print e    
   # return value must be iterable    
   return [HELLO_WORLD]