rocketmq-client-rust 0.9.0

Rust implementation of Apache rocketmq client
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

use criterion::criterion_group;
use criterion::criterion_main;
use criterion::Criterion;
use rocketmq_client_rust::producer::message_queue_selector::MessageQueueSelector;
use rocketmq_client_rust::producer::queue_selector::SelectMessageQueueByHash;
use rocketmq_client_rust::producer::queue_selector::SelectMessageQueueByRandom;
use rocketmq_common::common::message::message_queue::MessageQueue;
use rocketmq_common::common::message::message_single::Message;
use std::cell::RefCell;
use std::collections::hash_map::DefaultHasher;
use std::hash::Hash;
use std::hash::Hasher;
use std::hint::black_box;

thread_local! {
    static HASHER: RefCell<DefaultHasher> = RefCell::new(DefaultHasher::new());
}

/// Queue selector using thread-local hasher for performance comparison.
struct SelectMessageQueueByHashCached;

impl<M, A> MessageQueueSelector<M, A> for SelectMessageQueueByHashCached
where
    M: rocketmq_common::common::message::MessageTrait,
    A: Hash,
{
    #[inline]
    fn select(&self, mqs: &[MessageQueue], _msg: &M, arg: &A) -> Option<MessageQueue> {
        if mqs.is_empty() {
            return None;
        }

        let hash_code = HASHER.with(|hasher| {
            let mut h = hasher.borrow_mut();
            arg.hash(&mut *h);
            let code = h.finish();
            *h = DefaultHasher::new();
            code
        });

        let index = (hash_code % mqs.len() as u64) as usize;
        mqs.get(index).cloned()
    }
}

fn bench_hasher_creation(c: &mut Criterion) {
    c.bench_function("hasher_new", |b| {
        b.iter(|| {
            let hasher = black_box(DefaultHasher::new());
            black_box(hasher);
        })
    });
}

fn bench_select_original(c: &mut Criterion) {
    let selector = SelectMessageQueueByHash;
    let queues = vec![
        MessageQueue::from_parts("test_topic", "broker-a", 0),
        MessageQueue::from_parts("test_topic", "broker-a", 1),
        MessageQueue::from_parts("test_topic", "broker-a", 2),
        MessageQueue::from_parts("test_topic", "broker-a", 3),
    ];
    let msg = Message::builder().topic("test_topic").build().unwrap();

    c.bench_function("select_original", |b| {
        b.iter(|| {
            let order_id = black_box(12345);
            black_box(selector.select(&queues, &msg, &order_id))
        })
    });
}

fn bench_select_cached(c: &mut Criterion) {
    let selector = SelectMessageQueueByHashCached;
    let queues = vec![
        MessageQueue::from_parts("test_topic", "broker-a", 0),
        MessageQueue::from_parts("test_topic", "broker-a", 1),
        MessageQueue::from_parts("test_topic", "broker-a", 2),
        MessageQueue::from_parts("test_topic", "broker-a", 3),
    ];
    let msg = Message::builder().topic("test_topic").build().unwrap();

    c.bench_function("select_cached", |b| {
        b.iter(|| {
            let order_id = black_box(12345);
            black_box(selector.select(&queues, &msg, &order_id))
        })
    });
}

fn bench_select_random(c: &mut Criterion) {
    let selector = SelectMessageQueueByRandom;
    let queues = vec![
        MessageQueue::from_parts("test_topic", "broker-a", 0),
        MessageQueue::from_parts("test_topic", "broker-a", 1),
        MessageQueue::from_parts("test_topic", "broker-a", 2),
        MessageQueue::from_parts("test_topic", "broker-a", 3),
    ];
    let msg = Message::builder().topic("test_topic").build().unwrap();

    c.bench_function("select_random", |b| {
        b.iter(|| black_box(selector.select(&queues, &msg, &())))
    });
}

criterion_group!(
    benches,
    bench_hasher_creation,
    bench_select_original,
    bench_select_cached,
    bench_select_random
);
criterion_main!(benches);