为什么我有时会在使用 SQS 客户端时出现 Key Error

Question

我正在使用 boto3 SQS 客户端从 AWS SQS FIFO 队列接收消息。

def consume_msgs():
    sqs = None
    try:
        sqs = boto3.client('sqs',
                       region_name=S3_BUCKET_REGION,
                       aws_access_key_id=AWS_ACCESS_KEY_ID,
                       aws_secret_access_key=AWS_SECRET_ACCESS_KEY)
    except Exception:
        logger.warning('SQS client error {}'.format(sys.exc_info()[0]))
        logger.error(traceback.format_exc())

  ### more code to process message

该应用程序使用upstart. 大部分时间都可以正常工作。但有时当我在代码更改后重新启动服务时，应用程序会退出并显示以下错误

2018-10-06 01:29:38,654 WARNING SQS client error <class 'KeyError'>
2018-10-06 01:29:38,658 WARNING SQS client error <class 'KeyError'>
2018-10-06 01:29:38,663 ERROR Traceback (most recent call last):
  File "/home/ec2-user/aae_client/app/run.py", line 194, in consume_msgs
    aws_secret_access_key=AWS_SECRET_ACCESS_KEY)
  File "/home/ec2-user/aae_client/env/lib64/python3.6/dist-packages/boto3/__init__.py", line 83, in client
    return _get_default_session().client(*args, **kwargs)
  File "/home/ec2-user/aae_client/env/lib64/python3.6/dist-packages/boto3/session.py", line 263, in client
    aws_session_token=aws_session_token, config=config)
  File "/home/ec2-user/aae_client/env/lib64/python3.6/dist-packages/botocore/session.py", line 851, in create_client
    endpoint_resolver = self.get_component('endpoint_resolver')
  File "/home/ec2-user/aae_client/env/lib64/python3.6/dist-packages/botocore/session.py", line 726, in get_component
    return self._components.get_component(name)
  File "/home/ec2-user/aae_client/env/lib64/python3.6/dist-packages/botocore/session.py", line 926, in get_component
    del self._deferred[name]
KeyError: 'endpoint_resolver'

重新启动服务通常会修复它。每次重新启动服务时都不会发生这种情况。令人困惑的是KeyError导致实际错误回溯的警告。这究竟KeyError指的是什么？不可能是AWS_SECRET_ACCESS_KEY因为这个键永远不会改变，而且它在大多数情况下都可以正常工作。这个问题发生得很随机，来来去去。因此很难调试。我不明白这个错误是如何逃脱try..except块的

编辑

根据评论，这似乎与多线程有关。consume_msg确实由多个线程 def process_msgs() 运行：

for i in range(NUM_WORKERS):
    t = threading.Thread(target=consume_msgs, name='worker-%s' % i)
    t.setDaemon(True)
    t.start()
while True:
    time.sleep(MAIN_PROCESS_SLEEP_INTERVAL)

Answer 1

也许我误解了其他一些答案，但在多线程执行的情况下，如果这些函数在单独的线程中执行，我认为拥有一个 boto3 客户端对象并将其传递给其他函数不会起作用。endpoint_resolver我在调用 boto3 客户端服务时遇到了偶发错误，通过遵循文档中的示例以及 boto3 GitHub 问题（例如#1246和#1592 ）的评论，并在每个线程中创建单独的会话对象，这些错误被停止。就我而言，这意味着对我的代码进行几乎微不足道的更改，从

client = boto3.client(variant, region_name = creds['region_name'],
                      aws_access_key_id = ...,
                      aws_secret_access_key = ...)

到

session = boto3.session.Session()
client = session.client(variant, region_name = creds['region_name'],
                        aws_access_key_id = ...,
                        aws_secret_access_key = ...)

在单独线程中执行的函数中。我对OP代码的解读consume_msgs()是，可以进行类似的更改，并且可以消除偶尔的endpoint_resolver错误。

Answer 2

这个github问题建议你应该在顶层设置一次sqs客户端（而不是在函数中）：

sqs = boto3.client('sqs',
                   region_name=S3_BUCKET_REGION,
                   aws_access_key_id=AWS_ACCESS_KEY_ID,
                   aws_secret_access_key=AWS_SECRET_ACCESS_KEY)


def consume_msgs():
    # code to process message