Riak 是 Basho 公司推出的分布式 NoSQL 数据库，设计上参考了 Amazon Dynamo 论文中的一致性哈希、虚拟节点与向量时钟等机制。在 Python 环境中，官方提供了 riak-python-client 这个库，它封装了 Protocol Buffers 与 HTTP 接口，允许开发者用简洁的 API 完成键值存储、索引查询和集群状态读取。本文通过实际代码演示如何安装并连接 Riak、执行数据操作以及处理并发写入产生的冲突。
安装与基本连接配置
安装 riak-python-client 非常简单，可以直接使用 pip 命令。需要注意的是，如果希望通过 Protocol Buffers 协议通信，通常还需要安装对应的 protobuf 支持；如果只使用 HTTP 接口，则依赖更少。安装完成后，第一步是创建客户端实例并指定 Riak 节点地址。
from riak import RiakClient client = RiakClient(protocol='pbc', host='127.0.0.1', pb_port=8087) print(client.ping())
上面的代码使用 Protocol Buffers 协议连接本机 Riak 节点，默认端口为 8087。ping() 方法会返回 True 或 False，用于检测节点是否可用。如果集群部署了多个节点，也可以在初始化时传入 nodes 参数，客户端会按列表顺序尝试连接。建议把客户端实例创建为模块级单例或放入连接池，避免每次请求都重新建立 TCP 连接。
连接参数中比较重要的是协议选择与超时设置。HTTP 协议默认端口为 8098，适合防火墙限制较多的环境；Protocol Buffers 性能更好，适合生产环境使用。还可以通过 timeout 参数设置网络请求超时，防止某个节点无响应时拖垮调用线程。例如 RiakClient(protocol='pbc', nodes=[{'host':'192.168.1.10','pb_port':8087}], timeout=5)。
Bucket 操作与数据写入读取
Riak 的数据组织方式是 Bucket 加 Key。Bucket 可以理解为命名空间，Key 是其中的唯一标识。客户端提供了 bucket() 方法获取一个 Bucket 对象，然后通过 new() 创建数据对象并设置值，最后调用 store() 完成写入。
bucket = client.bucket('users')
user = bucket.new('user_001', data={
'name': 'alice',
'email': 'alice@ipipp.com',
'age': 28
})
user.store()
print(user.key)
读取数据时，使用 bucket.get() 方法，传入 Key 即可返回对象。对象上可以通过 data 属性拿到 Python 字典或字符串，还可以通过 exists 字段判断键是否存在。如果键不存在，返回对象的 data 为 None，不会抛出异常。
fetched = bucket.get('user_001')
if fetched.exists:
print(fetched.data)
else:
print('key not found')
如果需要删除数据，调用对象的 delete() 方法即可。Riak 默认采用最终一致性，写入成功后可能不会立即在所有副本可见。如果业务需要强一致读取，可以在请求参数中增加 r=2 之类的读取仲裁数，例如 bucket.get('user_001', r=2)，表示至少从两个副本读取一致结果。
二级索引与 MapReduce 查询
Riak 虽然以键值存储为主，但支持二级索引（2i），可以对对象附加索引字段。写入时在对象上添加 add_index() 方法，然后通过 get_index() 进行范围查询。二级索引适合按非主键字段检索，比如按用户年龄段查找。
user.add_index('age_bin', 20)
user.store()
results = bucket.get_index('age_bin', 20, 29)
for key in results:
print(key)
上面的示例中，age_bin 是索引名称，末尾以 _bin 表示二进制索引，_int 表示整数索引。范围查询时传入起始值与结束值，返回匹配的 Key 列表。需要注意的是，二级索引查询返回的是 Key，而不是完整对象，需要再通过 get() 获取数据。如果索引字段值变化，需要重新写入对象并更新索引。
对于更复杂的分析场景，Riak 提供了 MapReduce 框架。虽然官方建议优先使用二级索引或搜索组件，但了解 MapReduce 仍然有助于理解 Riak 的分布式计算能力。客户端可以提交 map 和 reduce 阶段的 JavaScript 或 Erlang 函数。下面是一个简单的 Map 阶段示例，提取所有对象的 email 字段。
query = client.add('users')
query.map("""
function(v) {
var data = JSON.parse(v.values[0].data);
return [data.email];
}
""")
for result in query.run():
print(result)
需要注意的是，MapReduce 查询会消耗较多集群资源，尤其在大数据集上执行时容易影响在线业务。生产环境中建议通过 Riak Search 或预先维护索引来替代复杂的 MapReduce 任务。
冲突处理与一致性策略
Riak 允许同一 Key 被并发更新，当不同节点同时写入时会产生冲突，系统会保留多个版本。读取时，客户端默认会根据向量时钟自动选择较新的版本，但如果无法判断先后，可能返回多个兄弟值（siblings）。这种情况下需要业务层进行合并。
obj = bucket.get('user_001')
if obj.siblings:
for sibling in obj.siblings:
print(sibling.data)
处理冲突的常见做法是在写入前使用 if_none_match 或依赖向量时钟进行条件更新。也可以在读取到 siblings 后按业务规则合并，例如以时间戳最新的字段为准，合并后再次 store。客户端支持开启 allow_mult 属性来控制是否允许多版本存在，如果关闭，Riak 会在写入时自动选择一种解决策略，但可能丢失部分更新。
还需要关注读写仲裁参数。Riak 中 r 表示读取成功所需副本数，w 表示写入成功所需副本数，dw 表示持久化写入副本数。通过调整这些参数，可以在可用性与一致性之间权衡。例如需要强一致写入时设置 w=3 并要求三个副本确认，但会降低可用性。客户端调用时可以在方法参数中覆盖这些值，例如 obj.store(w=2, dw=1)。
总体来看，riak-python-client 提供了清晰的接口来操作 Riak，但分布式数据库的特性决定了开发者必须理解最终一致性、仲裁与冲突解决机制。合理配置客户端连接、索引与一致性参数，才能在保证可用性的同时满足业务对数据正确性的要求。