#572 Matplotlib 基础
Python 数据可视化 Matplotlib 科学计算 2021-06-20常用的一个 Python 图表工具。
#571 Pandas 基础
Python 科学计算 Pandas 2021-06-19基于 NumPy 的一个数据分析工具。
#570 Python 源码学习 02: PyObject
Python PythonSourceCode 2021-06-19源码
Include/object.h
/* Nothing is actually declared to be a PyObject, but every pointer to
* a Python object can be cast to a PyObject*. This is inheritance built
* by hand. Similarly every pointer to a variable-size Python object can,
* in addition, be cast to PyVarObject*.
*/
typedef struct _object {
_PyObject_HEAD_EXTRA // 如果开启了 Py_TRACE_REFS 增加一个 _ob_next, _ob_prev
// 使 all live heap objects 组成一个双向链表
Py_ssize_t ob_refcnt; // 长整型
PyTypeObject *ob_type;
} PyObject;
/* Cast argument to PyObject* type. */
#define _PyObject_CAST(op) ((PyObject*)(op))
#define _PyObject_CAST_CONST(op) ((const PyObject*)(op))
typedef struct {
PyObject ob_base;
Py_ssize_t ob_size; /* Number of items in variable part */
} PyVarObject;
/* Cast argument to PyVarObject* type. */
#define _PyVarObject_CAST(op) ((PyVarObject*)(op))
#define Py_REFCNT(ob) (_PyObject_CAST(ob)->ob_refcnt)
#define Py_TYPE(ob) (_PyObject_CAST(ob)->ob_type)
#define Py_SIZE(ob) (_PyVarObject_CAST(ob)->ob_size)
PyObject
相当于所有 Python 对象的父类，包含类型，引用计数等信息。
注释说的很清楚，不会有直接声明的 PyObject 变量，只会有 PyObject* 指针，所有指向 Python 对象的指针都可以转换成 PyObject*。
PyVarObject
表示 ob_size 个 PyObject，也就是说 PyVarObject 是一个 PyObject 的容器。
#569 NumPy 基础
Python 科学计算 NumPy 2021-06-18NumPy 提供了一个高效的数据结构（数组/矩阵）及对应运算支持，据说效率和 C 接近，是 Python 科学计算生态的基础。
#568 Golang RabbitMQ
Golang RabbitMQ MQ 2021-06-18- 2014/07/11, RabbitMQ & AMQP
- 2015/02/12, Python RabbitMQ
- 2015/04/01, RabbitMQ 管理命令
- 2015/05/04, pika 01: 基础
- 2015/05/04, pika 02: BlockingConnection
- 2015/05/05, pika 03: SelectConnection
- 2015/05/06, pika 04: TornadoConnection
- 2016/02/12, Java RabbitMQ
- 2018/01/31, RabbitMQ 集群
- 2018/04/01, 面试题：RabbitMQ 相关
- 2018/10/02, pika: AsyncioConnection
- 2018/10/06, 使用 RabbitMQ 遇到的一个小问题
- 2019/01/12, 延迟队列
- 2020/10/03, 常见的消息队列
- 2021/06/18, Golang RabbitMQ
RabbitMQ 是啥就不说了，怎么安装部署也不说了，就记录一下 RabbitMQ 在 Golang 开发中的应用。
说明：采用 github.com/streadway/amqp 库。
func (ch *Channel) Publish(exchange, key string, mandatory, immediate bool, msg Publishing) error
func (ch *Channel) Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args Table) (<-chan Delivery, error)
生产者
生产者基本流程
生产者：连接
amqp.Dial->amqp.Connectionamqp.Connection.Channel->amqp.Channel
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil {
log.Fatalf("Failed to connect to RabbitMQ: %s", err)
}
ch, err := conn.Channel()
if err != nil {
log.Fatalf("Failed to open a channel: %s", err)
}
生产者：配置（可选）
事先把 MQ 配好就行，但从稳妥起见，还是在连接时加上比较好。
amqp.Channel.QueueDeclareamqp.Channel.ExchangeDeclareamqp.Channel.QueueBind
q, err := ch.QueueDeclare(
"hello", // 队列名称
true, // 持久化
false, // 自动删除
false, // 独占
false, // 等待服务器回应
nil, // 额外参数
)
if err != nil {
log.Fatalf("Failed to declare a queue: %s", err)
}
生产者：发送
amqp.Channel.Publish
err = ch.Publish(
"", // exchange
q.Name, // routing key
false, // mandatory
false, // immediate
amqp.Publishing{
ContentType: "text/plain",
Body: []byte("hello world"),
})
if err != nil {
log.Fatalf("Failed to publish a message: %s", err)
}
生产者：收尾
amqp.Connection.Closeamqp.Channel.Close
消费者
消费者基本流程
和生产者基本一致。只是调用的的是 chan.Consume 而不是 chan.Publish。
然后就是配置阶段，消费者只用关心队列在不在。
package main
import (
"context"
"encoding/json"
"log"
"time"
"github.com/streadway/amqp"
)
// Consumer RabbitMQ 消费者
type Consumer struct {
conn *amqp.Connection
channel *amqp.Channel
queue string
}
// NewConsumer 创建消费者
func NewConsumer(conn *amqp.Connection, ch *amqp.Channel, queueName string) (*Consumer, error) {
// 声明队列
_, err = DeclareQueue(ch, queueName)
if err != nil {
return nil, err
}
return &Consumer{
conn: conn,
channel: ch,
queue: queueName,
}, nil
}
// Consume 消费消息
func (c *Consumer) Consume(ctx context.Context, workerID int) error {
// 设置 QoS
err := c.channel.Qos(
1, // prefetch count
0, // prefetch size
false, // global
)
if err != nil {
return err
}
// 开始消费
msgs, err := c.channel.Consume(
c.queue, // queue
"", // consumer
false, // auto-ack
false, // exclusive
false, // no-local
false, // no-wait
nil, // args
)
if err != nil {
return err
}
log.Printf("消费者 %d 已启动，等待消息...", workerID)
for {
select {
case <-ctx.Done():
log.Printf("消费者 %d 停止", workerID)
return nil
case msg, ok := <-msgs:
if !ok {
log.Printf("消费者 %d 消息通道关闭", workerID)
return nil
}
// 处理消息
var message Message
if err := json.Unmarshal(msg.Body, &message); err != nil {
log.Printf("解析消息失败: %v", err)
msg.Nack(false, false) // 拒绝消息，不重新入队
continue
}
// 模拟处理时间
log.Printf("消费者 %d 处理消息: %+v", workerID, message)
time.Sleep(1 * time.Second)
// 确认消息
if err := msg.Ack(false); err != nil {
log.Printf("确认消息失败: %v", err)
}
}
}
}
// Close 关闭连接
func (c *Consumer) Close() {
if c.channel != nil {
c.channel.Close()
}
if c.conn != nil {
c.conn.Close()
}
}
复习：Exchange Types
- Fanout: 忽略 routingKey / bindingKey，将消息广播给所有绑定队列
- Header: 忽略 routingKey / bindingKey，根据消息和队列绑定时指定的 headers 匹配
- 支持精准匹配、数字匹配
- 匹配规则：x-match = all（默认）/ any
- 据说性能很差，没有使用过，官方文档都没有提供代码示例
- Direct: routingKey = bindingKey
- Topic: routingKey 模糊匹配 bindingKey（
*表示一个单词，#表示若干个单词）
默认 Exchange
- 名称：空字符串
- 类型：Direct
- 特性：所有已声明的队列都会隐式绑定（implicit route）到默认 Exchange，bindingKey queue.Name
所以 routingKey quene.Name 就可以将消息发送给对应的队列
参考资料与拓展阅读
#567 UUID
UUID 2021-06-17Universally Unique Identifier 通用唯一识别码
是 ISO/IEC 标准，也定义在 IETF 的 RFC4122 中。
128 位，也就是 16 字节，通常使用 32 位 16 进制数字，以 8-4-4-4-12 的形式表示，例如：d09abf7e-3e39-11ec-9dbc-b1755772e461
#566 NanoID 能不能取代 UUID
NanoID UUID 2021-06-17最近听说一种新的 ID 生成器，叫做 NanoID，很多地方那个都把它拿来和 UUID 做对比。
我的结论：NanoID 没啥了不起，就是个随机字符串，取代不了 UUID。
#565 为什么我不看好微服务
开发者 微服务 2021-06-14我反对的微服务，不是这种思想，而是大家炒热的各种技术框架。
#564 Python 源码学习 01: 初步了解
Python PythonSourceCode 2021-06-09开始看 Python 源码，啃一啃这块硬骨头，在学习 Python 底层的同时，也可以提升自己的 C 语言能力。
#563 Golang Redis
Golang Redis 2021-06-05package main
import (
"log"
"time"
"github.com/go-redis/redis"
)
func main() {
client := redis.NewClient(&redis.Options{
Addr: "localhost:6379",
Password: "",
DB: 0,
})
var err error
// 使用 Get/Set/Del =====================================
err = client.Set("key", "value", 0).Err()
if err != nil {
panic(err)
}
val, err := client.Get("key").Result()
if err != nil {
panic(err)
}
log.Println("key", val)
err = client.Del("key").Err()
if err != nil {
panic(err)
}
// 使用 Pipeline ========================================
pipeline := client.Pipeline()
pipeline.Set("key1", "value1", time.Minute*5)
pipeline.Set("key2", "value2", time.Minute*5)
pipeline.Set("key3", "value3", time.Minute*5)
_, err = pipeline.Exec()
if err != nil {
panic(err)
}
pipeline.Get("key1")
pipeline.Get("key2")
pipeline.Get("key3")
vals, err := pipeline.Exec()
if err != nil {
panic(err)
}
val1, _ := vals[0].(*redis.StringCmd).Result()
val2, _ := vals[1].(*redis.StringCmd).Result()
val3, _ := vals[2].(*redis.StringCmd).Result()
log.Println("key1", val1)
log.Println("key2", val2)
log.Println("key3", val3)
pipeline.Del("key1", "key2", "key3")
_, err = pipeline.Exec()
if err != nil {
panic(err)
}
}
package main
import (
"log"
"github.com/go-redis/redis"
)
func main() {
client := redis.NewClient(&redis.Options{
Addr: "localhost:6379",
Password: "",
DB: 0,
})
log.Println(client)
var err error
var items []string
// List ===============================================
{
err = client.RPush("mylist", "item1", "item2", "item3").Err()
if err != nil {
panic(err)
}
listLen, err := client.LLen("mylist").Result()
if err != nil {
panic(err)
}
log.Println("Length of mylist:", listLen)
items, err = client.LRange("mylist", 0, -1).Result()
if err != nil {
panic(err)
}
log.Println("===== Items in mylist:", items)
}
// Hash ===============================================
{
err = client.HSet("myhash", "field1", "value1").Err()
if err != nil {
panic(err)
}
value1, err := client.HGet("myhash", "field1").Result()
if err != nil {
panic(err)
}
log.Println("Value of field1:", value1)
allFields, err := client.HGetAll("myhash").Result()
if err != nil {
panic(err)
}
log.Println("===== All fields in myhash:", allFields)
}
// Set ================================================
{
err = client.SAdd("myset", "item1", "item2", "item3").Err()
if err != nil {
panic(err)
}
setLen, err := client.SCard("myset").Result()
if err != nil {
panic(err)
}
log.Println("Length of myset:", setLen)
items, err = client.SMembers("myset").Result()
if err != nil {
panic(err)
}
log.Println("===== Items in myset:", items)
}
// ZSet ===============================================
{
err = client.ZAdd("myzset", redis.Z{Score: 1.0, Member: "one"}, redis.Z{Score: 2.0, Member: "two"}).Err()
if err != nil {
panic(err)
}
setLen, err := client.ZCard("myzset").Result()
if err != nil {
panic(err)
}
log.Println("Length of myzset:", setLen)
items, err = client.ZRange("myzset", 0, -1).Result()
if err != nil {
panic(err)
}
log.Println("===== Items in myzset:", items)
}
}
哨兵
这样理论上来说，肯定会有性能损耗，毕竟增加了和哨兵的通信。
具体能差多少，还得实验。
package main
import (
"fmt"
"github.com/go-redis/redis/v8"
)
func main() {
failoverClient := redis.NewFailoverClient(&redis.FailoverOptions{
SentinelAddrs: []string{"sentinel1:26379", "sentinel2:26379", "sentinel3:26379"},
MasterName: "mymaster",
})
pong, err := failoverClient.Ping().Result()
if err != nil {
panic(err)
}
fmt.Println(pong)
}
集群
package main
import (
"fmt"
"github.com/go-redis/redis/v8"
)
func main() {
clusterClient := redis.NewClusterClient(&redis.ClusterOptions{
Addrs: []string{"redis1:6379", "redis2:6379", "redis3:6379"},
})
pong, err := clusterClient.Ping().Result()
if err != nil {
panic(err)
}
fmt.Println(pong)
}