标签: rabbitmq

我们的一台测试部署服务器上的磁盘已满。它通常运行多个 docker 容器，其中几个使用rabbitmq 进行通信（也作为容器运行，镜像rabbitmq:3.8-management）。运行后docker-compose down大约3Gb磁盘空间被释放，并docker-compose up -d成功启动一切。

在调查磁盘已满的原因时，最大的贡献者是 /var/lib/docker/volumes，它使用了 63Gb 磁盘空间。

root@...:/var/lib/docker/volumes# du -s * | sort -nr | head -20
10937880        ca7780bbddd7ce3e4cb9074fc9cad60b9c3b0bf184436ac8f0545fe513e09b31
777792  ff949d7a36874231e2cc9184efbd6a5dcbad1c5931dc97aa03f2353c07af662a
526308  ffc6a025921c6cd32336f402e5f6fc50c1e8bf1ff8c3ed0a145fbcc24d778954
524972  921e79b99d80c040968da6abd906125026d133b5ad790c455bf5912966586de6
518308  c4796d0a7866de9cdc2ae752217e17a20ff4670fdb5dc2538b11be3b9f2bc20a
517352  ef0667fbe6b3049f390ed9a32480515db1a6491d31ba8021cba351851e4177cd
516616  7a230b2ade824f92905c52b5e7ed7b29aafe685de4e5334fe893139e1266db81
515968  c81dbd0a64ae5b120a643d564e29a67a65eb0a8535c986c36872c0999ecc4d6c
514108  283b55310eba29bfa197d2cdacfb4198188df4e6a43e02ee4f0bba91a26b58e6
512200  3e761db0bec2102b1fb76a4fcfddca1ab5473d535c8060d95ac266901073db36
510316  ceef371f07df37fffe36297e13645bc45a934d7cace9008427b4320f95cbf757
509024  390ca3483aa0d7a7a78f83825fb3c6949e66f60aba40daa5dec224692e790a2f
503556  9b450cf79f63b65a1b0b5f91116c20b5fd2759c102c20df1740020598c24edcd
501164  cbdd8ef3500c5ddd3cca27a6037e3851f8698fb6b2df7b1a16c4272475101f1d
500784  e84d84ece86c2966649141f2eb362f66c9e29e33eb1eebdd1da8e59e8ab85b6e
500312  27935ab135130f336d4947c8a450213ea6ca18db0eab76ee46ed35cff7491e0e
491584  619edf2f5bf74795df75581ed8ed79f8c117363ffe2272f20dfc9c460345947a
491440  f6e6eaba7f584796c548c3313993f9b7999c8ad29389ac7577ebb4f24d53c8ce
488288  2f4a8295042d1abcb19cfc8f81d12ae1a0aba297c0ac3161d001c1bfb48e0ee2
488216  b73a38a4250b3e75c019b43f67afa78f2f101789b0696985f9029eccec763f0b
root@...:/var/lib/docker/volumes#

忽略两个最大的卷，剩余磁盘空间的大部分被各种 00000001.wal 文件占用：

root@...:/var/lib/docker/volumes# find . -name '*.wal' -print0 | xargs -0 ls -lSh …

我已经设置了 dlq 和 dlx，但失败的消息没有重定向到 dlq。我正在尝试从java应用程序以及从rabbitmq服务器发送消息到MESSAGES.EXCHANGE，在这两种情况下我都会收到消息，但在抛出异常消息后应该重定向到DLX.MESSAGES.EXCHANGE，但它正在发生。

下面是java代码和rabbitmq服务器的屏幕截图。一切对我来说看起来都不错。在代码或rabbitmq服务器中找不到任何问题。

队列设置代码-

public class DLQAmqpConfiguration {
    public static final String DLX_MESSAGES_EXCHANGE = "DLX.MESSAGES.EXCHANGE";
    public static final String DLQ_MESSAGES_QUEUE = "DLQ.MESSAGES.QUEUE";

    public static final String MESSAGES_QUEUE = "MESSAGES.QUEUE";
    public static final String MESSAGES_EXCHANGE = "MESSAGES.EXCHANGE";
    public static final String ROUTING_KEY_MESSAGES_QUEUE = "ROUTING_KEY_MESSAGES_QUEUE";

    @Bean
    Queue messagesQueue() {
        return QueueBuilder.durable(MESSAGES_QUEUE)
                .withArgument("x-dead-letter-exchange", DLX_MESSAGES_EXCHANGE)
                .build();
    }

    @Bean
    DirectExchange messagesExchange() {
        return new DirectExchange(MESSAGES_EXCHANGE);
    }

    @Bean
    Binding bindingMessages() {
        return BindingBuilder.bind(messagesQueue()).to(messagesExchange()).with(ROUTING_KEY_MESSAGES_QUEUE);
    }

    @Bean
    FanoutExchange deadLetterExchange() {
        return new FanoutExchange(DLX_MESSAGES_EXCHANGE);
    }

    @Bean …

我有一个队列 {QueueName}。我定义了消费者和错误消息消费者如下：

                        cfg.ReceiveEndpoint
                        (
                            queueName: QueueName,
                            e =>
                            {
                                 e.UseMessageRetry(r => r.Immediate(2));

                                e.AutoDelete = false;
                                e.Durable = true;
                                e.Consumer(() => container.Resolve<My_Consumer>());
                                e.Consumer(() => container.Resolve<My_Fault_Consumer>());
                            }
                        );
                    

当消费者耗尽处理消息的尝试次数时，故障消息消费者就会启动并通过记录错误来处理消息。我注意到创建了额外的队列，名为 {QueueName}_error。

My_Fault_Consumer 不确认故障消息消耗并且队列增长。

如何确认这些消息？

我是一名Python菜鸟，目前正在使用rabbitMQ。我从rabbitMQ 收到有效负载并对其进行处理。这是第一个函数

def verify_user_id(ch, method, properties, body):
    print("[*] Received message for NIN verification")
    body = json.loads(body)  # convert JSON string to python object
    parameters = body.get('parameters')
    id_info = {
        'first_name': parameters.get('first_name'),
        'last_name': parameters.get('last_name'),
        'country': parameters.get('country'),
        'id_type': parameters.get('id_type'),
        'id_number': parameters.get('id_number'),
        'entered': parameters.get('entered')
    }
    partner_params = {
        'job_id': parameters.get('job_id'),
        'user_id': parameters.get('user_id'),
        'job_type': parameters.get('job_type')
    }
    return body, id_info, partner_params

现在，我想使用上面的函数向服务器发送请求，如下所示：

def smile_identity_func():
    body, id_info, partner_params = verify_user_id()
    try:
      
         connection = IdApi(str(partner_id), str(api_key), sid_server)
         response = connection.submit_job(partner_params, id_info)
         obj = response.json()
         return jsonify(obj)
        
        ch.basic_ack(delivery_tag=method.delivery_tag) …

我从 RabbitMq 服务器收到此错误

通道被服务器关闭：406（PRECONDITION-FAILED），消息为“PRECONDITION_FAILED - 未知的交货标签 80”

发生这种情况是因为在消费者任务期间连接丢失，最后，当消息被确认/确认时，我收到此错误，因为我无法在与我收到消息的通道不同的通道上确认消息。

这是 RabbitMq 连接的代码

async connect({ prefetch = 1, queueName }) {
    this.queueName = queueName;
    console.log(`[AMQP][${this.queueName}] | connecting`);
    return queue
        .connect(this.config.rabbitmq.connstring)
        .then(conn => {
            conn.once('error', err => {
                this.channel = null;
                if (err.message !== 'Connection closing') {
                    console.error(
                        `[AMQP][${this.queueName}] (evt:error) | ${err.message}`,
                    );
                }
            });

            conn.once('close', () => {
                this.channel = null;
                console.error(
                    `[AMQP][${this.queueName}] (evt:close) | reconnecting`,
                );
                this.connect({ prefetch, queueName: this.queueName });
            });
            return conn.createChannel();
        })
        .then(ch => {
            console.log(`[AMQP-channel][${this.queueName}] created`);
            ch.on('error', …

我在线找到了在rabbitmq上添加/创建新用户的步骤，示例文档

但我的内部团队有一个特殊要求，允许用户仅访问虚拟主机下的某些特定服务/队列/

现在，我创建的用户可以访问虚拟主机/下的所有队列

假设，如果我们需要更改它们的权限以仅访问虚拟主机 / 下的特定队列/服务（例如 cart-order1、cart-order2、cart-order3），我们该怎么做？

我们从 Masstransit 7.0.6 升级到 8.0.3

之前我们正在配置ConfigureJsonSerializer. 但这似乎从 8.0.3 版本中删除了。源代码如下

x.UsingRabbitMq((context, cfg) =>
{
    cfg.Host(Config.GetValue<string>("RabbitMQ:Host"), "/", h =>
    {
        h.Username(Config.GetValue<string>("RabbitMQ:Username"));
        h.Password(Config.GetValue<string>("RabbitMQ:Password"));
    });

    cfg.ConfigureEndpoints(context);

    cfg.ConfigureJsonSerializer(settings => { settings.DefaultValueHandling = DefaultValueHandling.Include; return settings; });
});

我们如何替换ConfigureJsonSerializer和DefaultValueHandling？

我看到有

cfg.UseJsonSerializer();
cfg.ConfigureJsonSerializerOptions(options => options.DefaultIgnoreCondition = System.Text.Json.Serialization.JsonIgnoreCondition.???);

这是正确的替代品吗DefaultIgnoreCondition？

Zipkin 在本地运行良好，但在 docker 容器中运行不佳。所有的微服务都在Eureka中注册良好，并且可以很好地通信。但唯一的问题是Zipkin。我收到以下错误：

org.springframework.web.client.ResourceAccessException：“http://localhost:9411/api/v2/spans”的 POST 请求出现 I/O 错误：连接到 http://localhost:9411 [localhost/127.0.0.1]失败：连接被拒绝

我的 docker-compose.yaml 如下：

version: '3.8'

services:
  currency-exchange:
    image: samankt/springboot-udemy-currency-exchange:0.0.1-SNAPSHOT
    mem_limit: 512m
    ports:
      - '8000:8000'
    networks:
      - saman-network  
    environment:
      EUREKA.CLIENT.SERVICE-URL.DEFAULTZONE: http://naming-server:8761/eureka
      EUREKA.INSTANCE.PREFERIPADDRESS: true  
      SPRING.ZIPKIN.BASE-URL: http://zipkin-server:9411/
      RABBIT_URI: amqp://guest:guest@rabbitmq:5672
      SPRING_RABBITMQ_HOST: rabbitmq
      SPRING_ZIPKIN_SENDER_TYPE: rabbit
    depends_on:
      - naming-server
      - rabbitmq
      
  api-gateway:
    image: samankt/springboot-udemy-currency-api-gateway:0.0.1-snapshot
    mem_limit: 512m
    ports:
      - '8765:8765'
    networks:
      - saman-network  
    environment:
      EUREKA.CLIENT.SERVICE-URL.DEFAULTZONE: http://naming-server:8761/eureka
      EUREKA.INSTANCE.PREFERIPADDRESS: true  
      SPRING.ZIPKIN.BASE-URL: http://zipkin-server:9411/
      RABBIT_URI: amqp://guest:guest@rabbitmq:5672
      SPRING_RABBITMQ_HOST: rabbitmq
      SPRING_ZIPKIN_SENDER_TYPE: rabbit 
    depends_on:
      - naming-server
      - rabbitmq
      
  currency-converter:
    image: samankt/currency-conversion:0.0.1-SNAPSHOT
    mem_limit: 700m …

我对 RabbitMQ 比较陌生，发现它非常方便和快捷，我用它通过 ruby​​ + bunny gem 来通信小对象。

现在我尝试传递每个大约 10~20MB 的对象进行交换，并将其扇出给其订阅者。

看起来工作正常，但是使用 RabbitMQ 作为发布者是一个好的做法吗？或者我应该对 RabbitMQ 使用一些猜想？

我是一个长期使用 Go 的 python 用户，我仍然有一些问题需要重新获得管理打字和指针的基本技能。

我有一个程序从 RabbitMq 接收事件（但无论我们在谈论什么传输，问题都是一样的）。其中一个甚至包含一个类型为 int 的可选字段 F1。

我的理解是，如果该字段在事件中不存在，那么 go 会将其默认为 0。但是 0 是该字段的有效值，我需要区分值为 0 的情况和值为 0 的情况未定义。

我想将我的字段设为 *int 以实际将“nil”作为值。但是当接收到一个事件时，F1 会被设置为实际指向的值，还是来自发送者的值地址？我还有其他选择吗？