阿里云ECS集群服务器时间同步解决方案

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了阿里云ECS集群服务器时间同步解决方案相关的知识,希望对你有一定的参考价值。

参考技术A 阿里云ECS实例已经提供了NTP服务器支持,直接启动已配置好的NTP服务即可。
《阿里云NTP服务器》
《配置Linux实例NTP服务》

在开启服务前,先确保环境配置:

文档 《配置Linux实例NTP服务》 中介绍了CentOS环境下开启NTP服务。
由于本人购买的Ubuntu服务器,下面总结Ubuntu环境下的配置。
执行命令查询所有服务,看ntp服务是否已开启(+号:已开启;-号:未开启):

发现香港地区的服务默认都没有开启ntp服务;但深圳地区的服务器默认已经开启了ntp服务。
执行命令查询ntp进程,发现深圳服务器默认已经开启了ntp服务:

如果未开启ntp服务,执行命令开启ntp服务:

开启成功后,如图:

或者查询ntp相关的进程:

重启后通过如下命令观察NTP的运行状态:

这个命令可以列出目前我们的 NTP 与相关的上层 NTP 的状态,上头的几个字段的意义为:

driftfile /etc/ntp/drift

语法为: restrict IP地址 mask 子网掩码 参数
其中IP地址也可以是default ,default 就是指所有的IP

参考 《ubuntu安装和使用NTP》

记录一次在阿里云ECS服务器部署验证RocketMQ的经历

==背景==

购买了3台阿里云ECS服务器,上面部署了RocketMQ,用来作为业务后台与平台之间的数据通讯中间件。

部署倒是异常顺利,不过在本地写程序,测试生产和消费数据的时候,出现了一些问题。

耗费了将近1天的时间,终于解决了,记录一下本次排查的经历。

 

==环境==

Linux:CentOS8(阿里云ECS服务器)

RocketMQ:4.6.1

 

==集群==

节点数:3个

节点1:broker-a(master)

节点2:broker-a(slave),broker-b(master)

节点3:broker-b(slave)

 

配置文件如下(IP地址省略了):

broker-a.properties

brokerClusterName=rexel
brokerName=broker-a
brokerId=0
deleteWhen=04
fileReservedTime=48
brokerRole=SYNC_MASTER
flushDiskType=ASYNC_FLUSH
listenPort=10921
brokerIP1=xx.xx.xx.01
namesrvAddr=xx.xx.xx.01:9876;xx.xx.xx.02:9876
autoCreateTopicEnable=false
autoCreateSubscriptionGroup=true
storePathRootDir=/home/radmin/data/rocketmq/rootdir-a-m
storePathCommitLog=/home/radmin/data/rocketmq/commitlog-a-m
storePathConsumerQueue=/home/radmin/data/rocketmq/consumequeue-a-m
storePathIndex=/home/radmin/data/rocketmq/index-a-m
storeCheckpoint=/home/radmin/data/rocketmq/checkpoint-a-m

 

broker-a-s.properties

brokerClusterName=rexel
brokerName=broker-a
brokerId=1
deleteWhen=04
fileReservedTime=48
brokerRole=SLAVE
flushDiskType=ASYNC_FLUSH
listenPort=10931
brokerIP1=xx.xx.xx.02
namesrvAddr=xx.xx.xx.01:9876;xx.xx.xx.02:9876
defaultTopicQueueNums=4
autoCreateTopicEnable=false
autoCreateSubscriptionGroup=true
storePathRootDir=/home/radmin/data/rocketmq/rootdir-a-s
storePathCommitLog=/home/radmin/data/rocketmq/commitlog-a-s
storePathConsumerQueue=/home/radmin/data/rocketmq/consumequeue-a-s
storePathIndex=/home/radmin/data/rocketmq/index-a-s
storeCheckpoint=/home/radmin/data/rocketmq/checkpoint-a-s

 

broker-b.properties

brokerClusterName=rexel
brokerName=broker-b
brokerId=0
deleteWhen=04
fileReservedTime=48
brokerRole=SYNC_MASTER
flushDiskType=ASYNC_FLUSH
listenPort=10921
brokerIP1=xx.xx.xx.02
namesrvAddr=xx.xx.xx.01:9876;xx.xx.xx.02:9876
defaultTopicQueueNums=4
autoCreateTopicEnable=false
autoCreateSubscriptionGroup=true
storePathRootDir=/home/radmin/data/rocketmq/rootdir-b-m
storePathCommitLog=/home/radmin/data/rocketmq/commitlog-b-m
storePathConsumerQueue=/home/radmin/data/rocketmq/consumequeue-b-m
storePathIndex=/home/radmin/data/rocketmq/index-b-m
storeCheckpoint=/home/radmin/data/rocketmq/checkpoint-b-m

 

broker-b-s.properties

brokerClusterName=rexel
brokerName=broker-b
brokerId=1
deleteWhen=04
fileReservedTime=48
brokerRole=SLAVE
flushDiskType=ASYNC_FLUSH
listenPort=10931
brokerIP1=xx.xx.xx.03
namesrvAddr=xx.xx.xx.01:9876;xx.xx.xx.02:9876
defaultTopicQueueNums=4
autoCreateTopicEnable=false
autoCreateSubscriptionGroup=true
storePathRootDir=/home/radmin/data/rocketmq/rootdir-b-s
storePathCommitLog=/home/radmin/data/rocketmq/commitlog-b-s
storePathConsumerQueue=/home/radmin/data/rocketmq/consumequeue-b-s
storePathIndex=/home/radmin/data/rocketmq/index-b-s
storeCheckpoint=/home/radmin/data/rocketmq/checkpoint-b-s

 

==最终代码== 

RocketUtils.java

package com.rexel.stream.common.utils;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.exception.RemotingException;

public class RocketUtils implements Serializable{
    private static RocketUtils rocketUtils = null;
    private static Map<String, DefaultMQProducer> nameSrvMap = null;

    private RocketUtils() {

    }

    public synchronized static RocketUtils getInstance() {
        if (rocketUtils == null) {
            synchronized (RocketUtils.class) {
                rocketUtils = new RocketUtils();
            }
        }
        nameSrvMap = new HashMap<>();
        return rocketUtils;
    }

    public DefaultMQPushConsumer createConsumer(String namesrvAddr, String topic, String group) {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setInstanceName(UUID.randomUUID().toString());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.setVipChannelEnabled(false);
        consumer.setConsumeThreadMax(1);
        consumer.setConsumeThreadMin(1);
        consumer.setConsumeMessageBatchMaxSize(1);
        try {
            consumer.subscribe(topic, "*");
        } catch (MQClientException e) {
            e.printStackTrace();
            return null;
        }
        return consumer;
    }

    public DefaultMQProducer createProducer(String nameSrvAddr, String group) {
        if (nameSrvMap == null) {
            return null;
        }

        if (nameSrvMap.containsKey(nameSrvAddr)) {
            return nameSrvMap.get(nameSrvAddr);
        }

        DefaultMQProducer producer = new DefaultMQProducer(group);
        producer.setNamesrvAddr(nameSrvAddr);
        producer.setSendMessageWithVIPChannel(false);
        producer.setSendMsgTimeout(5000);
        producer.setInstanceName(UUID.randomUUID().toString());
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
            return null;
        }

        nameSrvMap.put(nameSrvAddr, producer);
        return producer;
    }

    public boolean sendOr(DefaultMQProducer producer, Message msg, boolean async) {
        if (async) {
            return sendAsync(producer, msg);
        } else {
            return send(producer, msg);
        }
    }

    public boolean sendAsync(DefaultMQProducer producer, Message msg) {
        try {
            producer.send(msg, new CallBack());
            return true;
        } catch (MQClientException | RemotingException | InterruptedException e) {
            e.printStackTrace();
            return false;
        }
    }

    public boolean send(DefaultMQProducer producer, Message msg) {
        try {
            producer.send(msg);
            return true;
        } catch (MQClientException | RemotingException | InterruptedException | MQBrokerException e) {
            e.printStackTrace();
            return false;
        }
    }

    private class CallBack implements SendCallback,Serializable{
        @Override
        public void onSuccess(SendResult sendResult) {
            System.out.println("[------]onSuccess");
        }

        @Override
        public void onException(Throwable throwable) {
            System.out.println("[------]onException. " + throwable.getMessage());
        }
    }
}

 

RmqProducer.java

package com.rexel.stream.tools;

import com.alibaba.fastjson.JSONObject;
import com.rexel.stream.common.utils.RocketUtils;
import java.nio.charset.StandardCharsets;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;

public class RmqProducer {
    public static void main(String[] args) {
        System.out.println("[------]start.");
        RocketUtils rocketUtils = RocketUtils.getInstance();
        DefaultMQProducer producer =
            rocketUtils.createProducer("xx.xx.xx.01:9876;xx.xx.xx.02:9876", "pro_test3");

        for (int i = 0; i < 10; i++) {
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("name", "VA_2YC_VAL");
            jsonObject.put("judge", "≥");
            jsonObject.put("value", "100");
            rocketUtils.sendAsync(producer, new Message(
                "app_notice",
                jsonObject.toJSONString().getBytes(StandardCharsets.UTF_8)));
        }

        //如果使用异步发送,这里不要shutdown
//        producer.shutdown();
        System.out.println("[------]end.");
    }
}

 

RmqConsumer.java

package com.rexel.stream.tools;

import com.rexel.stream.common.utils.RocketUtils;
import java.nio.charset.StandardCharsets;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.Message;

public class RmqConsumer {
    public static void main(String[] args) throws MQClientException {
        System.out.println("[------]start.");
        RocketUtils rocketUtils = RocketUtils.getInstance();
        DefaultMQPushConsumer consumer = rocketUtils.createConsumer(
            "xx.xx.xx.01:9876;xx.xx.xx.02:9876",
            "app_notice",
            "rexel_stream3");
        consumer.registerMessageListener(
            (MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
                for (Message msg : list) {
                    try {
                        byte[] body = msg.getBody();
                        String message = new String(body, StandardCharsets.UTF_8);
                        System.out.println("[------]rmq message= " + message);
                    } catch (Exception e) {
                        e.printStackTrace();
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
        consumer.start();

        System.out.println("[------]end.");
    }
}

 

==问题1==

配置完成之后,尝试在客户端编写生产者代码,结果生产数据的时候报错。

org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to [xxxx] failed

 

尝试1:

把生产者和消费者的代码中增加setSendMessageWithVIPChannel(false)。

结果:依然报错,错误没有改变

 

尝试2:

 

在配置文件中增加brokerIP1=xx.xx.xx.xx的配置。

结果:依然报错,错误没有改变

 

尝试3:

网上说是防火墙的问题,服务器本身的防火墙很早就已经被关闭了。尝试去设置阿里云ECS服务器产品的端口。

一次性的把一个10900/10999的端口全部开放

结果:测试同步生产数据正常。

 

尝试4:

测试异步生产数据。调用RocketUtils中的sendAsync方法。结果报错:

[------]onException. org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to [101.132.242.90:9876, 47.116.50.192:9876] failed
[------]onException. The producer service state not OK, SHUTDOWN_ALREADY
See http://rocketmq.apache.org/docs/faq/ for further details.
[------]onException. The producer service state not OK, SHUTDOWN_ALREADY
See http://rocketmq.apache.org/docs/faq/ for further details.
[------]onException. The producer service state not OK, SHUTDOWN_ALREADY
See http://rocketmq.apache.org/docs/faq/ for further details.
[------]onException. The producer service state not OK, SHUTDOWN_ALREADY
See http://rocketmq.apache.org/docs/faq/ for further details.
[------]onException. The producer service state not OK, SHUTDOWN_ALREADY
See http://rocketmq.apache.org/docs/faq/ for further details.
[------]onException. The producer service state not OK, SHUTDOWN_ALREADY
See http://rocketmq.apache.org/docs/faq/ for further details.
[------]onException. org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to [101.132.242.90:9876, 47.116.50.192:9876] failed
[------]onException. org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to [101.132.242.90:9876, 47.116.50.192:9876] failed
[------]onException. org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to [101.132.242.90:9876, 47.116.50.192:9876] failed

 

原因是使用异步生产数据的时候,我的程序里调用了shutdown方法,

导致后续的异步线程无法正常执行。注释掉shutdown处理之后,异步生产正常。

 

结论:

如果出现connect to [xxxx] failed的问题,不外乎尝试以下几种办法:

1、程序中:生产者或者消费者:setSendMessageWithVIPChannel(false)

2、配置文件:如果是阿里云ECS服务器,以下两个配置使用外网地址:

brokerIP1=xx.xx.xx.01
namesrvAddr=xx.xx.xx.01:9876;xx.xx.xx.02:9876

3、防火墙:关闭服务器本身的防火墙。

4、安全组:阿里云服务器本身的网络安全组中需要开通端口。

 

==问题2==

生产者已经没有问题了,但是消费者一直消费不到数据。程序不报任何错误,就是消费不到数据。

在网上找了一些到有的博客,其中这篇给了我一些方向,

http://www.jiangxinlingdu.com/rocketmq/2019/08/06/noconsumer.html

初步怀疑是消费者的偏移量有问题。

 

解决办法:

我这个环境由于是新搭环境,目前还不是生产环境,所以我直接采用的方式是:

1、停止rocketmq集群

2、删除所有rocketmq的文件

3、重启集群

4、重新创建topic

 

一套暴利连招之后,消费者果然可以消费到数据了。

索然没有真正的找到问题的原因,不过基本上可以确定是rocketmq的元数据出现了问题,

这个问题的产生可能是我最近不断的调试配置文件,修改内外网地址,重启引起的。

 

--END--

以上是关于阿里云ECS集群服务器时间同步解决方案的主要内容,如果未能解决你的问题,请参考以下文章

阿里云ECS服务器部署HADOOP集群:ZooKeeper 完全分布式集群搭建

阿里云ECS服务器部署HADOOP集群:Flume 安装

阿里云ECS服务器部署HADOOP集群:Pig 安装

阿里云ECS服务器部署HADOOP集群:Hive本地模式的安装

阿里云ECS服务器部署HADOOP集群:HBase完全分布式集群搭建(使用外置ZooKeeper)

记录一次在阿里云ECS服务器部署验证RocketMQ的经历