作者 | 计缘

众所周知，游戏行业在当今的互联网行业中算是一棵常青树。在疫情之前的 2019 年，中国游戏市场营收规模约 2884.8 亿元，同比增长 17.1%。2020 年因为疫情，游戏行业更是突飞猛进。玩游戏本就是中国网民最普遍的娱乐方式之一，疫情期间更甚。据不完全统计，截至 2019 年，中国移动游戏用户规模约 6.6 亿人，占中国总网民规模 8.47 亿的 77.92%，可见游戏作为一种低门槛、低成本的娱乐手段，已成为大部分人生活中习以为常的一部分。

对于玩家而言，市面上的游戏数量多如牛毛，那么玩家如何能发现和认知到一款游戏，并且持续的玩下去恐怕是所有游戏厂商需要思考的问题。加之 2018 年游戏版号停发事件，游戏厂商更加珍惜每一个已获得版号的游戏产品，所以这也使得“深度打磨产品质量”和“提高运营精细程度”这两个游戏产业发展方向成为广大游戏厂商的发展思路，无论是新游戏还是老游戏都在努力落实这两点：

• 新游戏：面向玩家需要提供更充足的推广资源和更完整的游戏内容。

• 老游戏：通过用户行为分析，投入更多的精力和成本，制作更优质的版本内容。

• 游戏研发商：研发游戏的公司，生产和制作游戏内容。比如王者荣耀的所有英雄设计、游戏战斗场景、战斗逻辑等，全部由游戏研发公司提供。

• 游戏发行商：游戏发行商的主要工作分三大块：市场工作、运营工作、客服工作。游戏发行商把控游戏命脉，市场工作核心是导入玩家，运营工作核心是将用户价值最大化、赚取更多利益。

• 游戏平台/渠道商：游戏平台和渠道商的核心目的就是曝光游戏，让尽量多的人能发现你的游戏。

• 当流量脉冲来的时候，这部分是否可以快速扩容以应对流量冲击。

• 游戏运营具备潮汐特性，并非天天都在进行，这就需要考虑如何优化资源利用率。

• 流量太大，节点加少了，导致一部分流量的数据没有采集到。

• 流量没有预期那么大，节点加多了，导致资源浪费。

• 无需运维介入，研发同学就可以很快的搭建出来。

• 无论流量大小，均可以平稳的承接。

• 函数计算拉起的实例数量可以紧贴流量大小的曲线，做到资源利用率最优化，再加上按量计费的模式，可以最大程度优化成本。

• 函数类型：HTTP（即触发器为 HTTP）

• 函数名称：receiveData

• 运行环境：Python3

• 函数实例类型：弹性实例

• 函数执行内存：512MB

• 函数运行超时时间：60 秒

• 函数单实例并发度：1

• 触发器类型：HTTP 触发器

• 触发器名称：defaultTrigger

• 认证方式：anonymous（即无需认证）

• 请求方式：GET，POST

# -*- coding: utf-8 -*-import loggingimport jsonimport urllib.parseHELLO_WORLD = b'Hello world!\n'def handler(environ, start_response):    logger = logging.getLogger()     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    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)      request_body_str = urllib.parse.unquote(request_body.decode("GBK"))    request_body_obj = json.loads(request_body_str)    logger.info(request_body_obj["action"])    logger.info(request_body_obj["articleAuthorId"])
    status = '200 OK'    response_headers = [('Content-type', 'text/plain')]    start_response(status, response_headers)    return [HELLO_WORLD]

sudo npm install @alicloud/fun -g

$ fun --version3.6.20

ROSTemplateFormatVersion: '2015-09-01'Transform: 'Aliyun::Serverless-2018-04-03'Resources:FCBigDataDemo:Type: 'Aliyun::Serverless::Service'Properties:Description: 'local invoke demo'VpcConfig:VpcId: 'vpc-xxxxxxxxxxx'VSwitchIds: [ 'vsw-xxxxxxxxxx' ]SecurityGroupId: 'sg-xxxxxxxxx'LogConfig:Project: fcdemoLogstore: fc_demo_storedataToKafka:Type: 'Aliyun::Serverless::Function'Properties:Initializer: index.my_initializerHandler: index.handlerCodeUri: './'Description: ''Runtime: python3

• FCBigDataDemo：自定义的服务名称。通过下面的 Type 属性标明是服务，即 Aliyun::Serverless::Service。

• Properties：Properties 下的属性都是该服务的各配置项。

• VpcConfig：服务的 VPC 配置，包含：

• VpcId：VPC ID。

• VSwitchIds：交换机 ID，这里是数组，可以配置多个交换机。

• SecurityGroupId：安全组 ID。

• LogConfig：服务绑定的日志服务（SLS）配置，包含：

• Project：日志服务项目。

• Logstore：LogStore 名称。

• dataToKafka：该服务下自定义的函数名称。通过下面的 Type 属性标明是函数，即 Aliyun::Serverless::Function。
  

• Properties：Properties下的属性都是该函数的各配置项。

• Initializer：配置初始化函数。

• Handler：配置入口函数。

• Runtime：函数运行环境。

fun install --runtime python3 --package-type pip kafka-python

# -*- coding: utf-8 -*-import loggingimport jsonimport urllib.parsefrom kafka import KafkaProducerproducer = Nonedef my_initializer(context):        logger = logging.getLogger()     logger.info("init kafka producer")    global producer    producer = KafkaProducer(bootstrap_servers='XX.XX.XX.XX:9092,XX.XX.XX.XX:9092,XX.XX.XX.XX:9092')def handler(event, context):    logger = logging.getLogger()       # 接收回传的数据    event_str = json.loads(event)    event_obj = json.loads(event_str)    logger.info(event_obj["action"])    logger.info(event_obj["articleAuthorId"])    # 向Kafka发送消息    global producer    producer.send('ikf-demo', json.dumps(event_str).encode('utf-8'))    producer.close()    return 'hello world'

• my_initializer：函数实例被拉起时会先执行该函数，然后再执行 handler 函数 ，当函数实例在运行时，之后的请求都不会执行 my_initializer 函数 。一般用于各种连接的初始化工作，这里将初始化 Kafka Producer 的方法放在了这里，避免反复初始化 Produer。

•  handler：该函数只有两个逻辑，接收回传的数据和将数据发送至 Kafka 的指定 Topic。

• 下面通过 fun deploy 命令部署函数，该命令会做两件事：
• 根据 template.yml 中的配置创建服务和函数。
• 将 index.py 和 .fun 上传至函数中。

# -*- coding: utf-8 -*-import loggingimport jsonimport urllib.parseimport fc2HELLO_WORLD = b'Hello world!\n'client = Nonedef my_initializer(context):        logger = logging.getLogger()     logger.info("init fc client")    global client    client = fc2.Client(        endpoint="http://your_account_id.cn-hangzhou-internal.fc.aliyuncs.com",        accessKeyID="your_ak",        accessKeySecret="your_sk"    )def handler(environ, start_response):    logger = logging.getLogger()     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    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)      request_body_str = urllib.parse.unquote(request_body.decode("GBK"))    request_body_obj = json.loads(request_body_str)    logger.info(request_body_obj["action"])    logger.info(request_body_obj["articleAuthorId"])    global client    client.invoke_function(        'FCBigDataDemo',        'dataToKafka',        payload=json.dumps(request_body_str),        headers = {'x-fc-invocation-type': 'Async'}    )
    status = '200 OK'    response_headers = [('Content-type', 'text/plain')]    start_response(status, response_headers)    return [HELLO_WORLD]

• 导入函数计算的库：import fc2

• 添加初始化方法，用于创建函数计算 Client：

def my_initializer(context):        logger = logging.getLogger()        logger.info("init fc client")        global client        client = fc2.Client(            endpoint="http://your_account_id.cn-hangzhou-internal.fc.aliyuncs.com",            accessKeyID="your_ak",            accessKeySecret="your_sk")

• 通过函数计算 Client 调用第二个函数

global client    client.invoke_function(            'FCBigDataDemo',            'dataToKafka',          payload=json.dumps(request_body_str),            headers = {'x-fc-invocation-type': 'Async'})

• 第一个参数：调用函数所在的服务名称。

• 第二个参数：调用函数的函数名称。

• 第三个参数：向调用函数传的数据。

• 第四个参数：调用第二个函数 Request Header 信息。这里主要通过 x-fc-invocation-type 这个 Key 来设置是同步调用还是异步调用。这里设置 Async 为异步调用。

4. 配置 Kafka

....producer = KafkaProducer(bootstrap_servers='XX.XX.XX.XX:9092,XX.XX.XX.XX:9092,XX.XX.XX.XX:9092')....

...# 第一个参数为Topic名称producer.send('ikf-demo', json.dumps(event_str).encode('utf-8'))...

final ParameterTool parameterTool = ParameterTool.fromArgs(args);String kafkaTopic = parameterTool.get("kafka-topic","ikf-demo");String brokers = parameterTool.get("brokers", "XX.XX.XX.XX:9092,XX.XX.XX.XX:9092,XX.XX.XX.XX:9092");Properties kafkaProps = new Properties();kafkaProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);kafkaProps.put(ConsumerConfig.GROUP_ID_CONFIG, "ikf-demo");FlinkKafkaConsumer<UserBehaviorEvent> kafka = new FlinkKafkaConsumer<>(kafkaTopic, new UserBehaviorEventSchema(), kafkaProps);kafka.setStartFromLatest();kafka.setCommitOffsetsOnCheckpoints(false);StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<UserBehaviorEvent> dataStreamByEventTime = env.addSource(kafka);

• 压力模式：

• 并发模式：指定有多少并发用户同时发请求。

• RPS模式：指定每秒有多少请求数。

• 递增模式：在压测过程中可以通过手动调节压力，也可以自动按百分比递增压力。

• 最大并发：同时有多少个虚拟用户发起请求。

• 递增百分比：如果是自动递增的话，按这里的百分比递增。

• 单量级持续时长：在未完全达到压力全量的时候，每一级梯度的压力保持的时长。

• 压测总时长：一共需要压测的时长。