在目标检测任务中，数据集的划分通常分为训练集和验证集，以便在训练模型时评估模型的性能。这个过程对于有效训练和评估目标检测模型非常重要。下面是划分目标检测数据集的一般步骤：
``

1. 数据集组织：
   确保你的数据集中包含图像文件（通常为 ".jpg" 或 ".png"）和相应的标签文件，用于描述图像中的目标位置和类别。标签文件可以是类似于 YOLO 格式的 ".txt" 文件，也可以是其他目标检测任务中使用的格式。

2. 划分比例选择：
   确定训练集和验证集的划分比例。通常，常见的比例为 80-90% 的数据用于训练，而剩下的 10-20% 用于验证。这个比例可以根据你的数据量和任务的特性进行调整。

3. 数据集划分代码：
   编写代码来将数据集划分为训练集和验证集。这通常涉及将图像文件和相应的标签文件分别移动到两个不同的文件夹中。

4. 文件夹结构：
   为了方便管理，创建一个包含两个子文件夹（例如 "images" 和 "labels"）的文件夹，分别用于存储图像和标签文件。训练集和验证集各自都应该有这样的文件夹结构。

5. 随机化和保持一一对应：
   在划分数据集之前，通常会对数据集进行随机化处理，以确保训练集和验证集中的样本都是随机选择的。在划分过程中，要保持图像和标签的一一对应关系，以确保训练和验证的一致性。

6. 数据集划分结果验证：
   验证划分的结果是否符合预期。你可以检查训练集和验证集的文件数量，确保图像和标签仍然保持一一对应，并且确保文件夹结构正确。
   ！！！ 注意以下代码中all_data应含有图片与对应标签（在同一个文件夹中，且名称相同仅拓展名不同，分为.jpg和.txt文件）

import os
import random
import shutil

def split_dataset(srcDir, trainDir, valDir, split_ratio=0.9):
    """
    将数据集划分为训练集和验证集，并保存到相应的文件夹中。

    Parameters:
    - srcDir: 原始数据集文件夹路径，包含图像和标签文件。
    - trainDir: 训练集文件夹路径，包含 'images' 和 'labels' 子文件夹。
    - valDir: 验证集文件夹路径，包含 'images' 和 'labels' 子文件夹。
    - split_ratio: 数据集划分比例，默认为 0.9，表示将 90% 的数据用于训练集，10% 用于验证集。
    """
    os.makedirs(os.path.join(trainDir, 'images'), exist_ok=True)
    os.makedirs(os.path.join(trainDir, 'labels'), exist_ok=True)
    os.makedirs(os.path.join(valDir, 'images'), exist_ok=True)
    os.makedirs(os.path.join(valDir, 'labels'), exist_ok=True)

    # 获取数据集中所有文件的列表
    file_list = os.listdir(srcDir)
    random.shuffle(file_list)

    # 根据划分比例计算训练集和验证集的边界索引
    split_index = int(len(file_list) * split_ratio)
    train_files = file_list[:split_index]
    val_files = file_list[split_index:]

    # 将训练集数据移动到相应文件夹
    for file in train_files:
        if file.endswith('.jpg'):
            img_src = os.path.join(srcDir, file)
            label_src = os.path.join(srcDir, file[:-4] + '.txt')
            shutil.move(img_src, os.path.join(trainDir, 'images', file))
            shutil.move(label_src, os.path.join(trainDir, 'labels', file[:-4] + '.txt'))

    # 将验证集数据移动到相应文件夹
    for file in val_files:
        if file.endswith('.jpg'):
            img_src = os.path.join(srcDir, file)
            label_src = os.path.join(srcDir, file[:-4] + '.txt')
            shutil.move(img_src, os.path.join(valDir, 'images', file))
            shutil.move(label_src, os.path.join(valDir, 'labels', file[:-4] + '.txt'))

if __name__ == '__main__':
    # 输入文件夹路径
    srcDir = r"C:\Users\86159\Desktop\hat\all_data"  
    trainDir = r'C:\Users\86159\Desktop\hat\train'  
    valDir = r'C:\Users\86159\Desktop\hat\val'  

    # 调用函数划分数据集
    split_dataset(srcDir, trainDir, valDir)

RocketMQ—RocketMQ发送同步、异步、单向、延迟、批量、顺序、批量消息、带标签消息

发送同步消息

生产者发送消息，mq进行确认，然后返回给生产者状态。这就是同步消息。

前文demo程序就是发送的同步消息。

发送异步消息

异步消息通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待Broker的响应。发送完以后会有一个异步消息通知。

代码如下：

/**
     * 异步消息测试
     */
@Test
public void simpleAsyncProducer() throws Exception {
    //创建一个生产者，并指定一个组名
    DefaultMQProducer producer = new DefaultMQProducer("async-producer-group");
    //连接namesrv,参数是namesrv的ip地址:端口号
    producer.setNamesrvAddr(MqConstant.NAME_SERVER_ADDR);
    //启动
    producer.start();


    //指定topic，创建一个消息
    Message message = new Message("asyncTopic1", "这是一条异步消息".getBytes());

    //发送异步消息，并设置回调内容
    producer.send(message, new SendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
            log.info("回调内容，发送成功");
        }

        @Override
        public void onException(Throwable throwable) {
            log.info("回调内容，发送失败");
        }
    });


    log.info("主线程执行中=========");

    System.in.read();
}

从运行结果可以看到是不同的线程输出的内容。

发送单向消息

这种方式主要用在不关心发送结果的场景，这种方式吞吐量很大，但是存在消息丢失的风险，例如日志信息的发送。

代码如下:

@Test
public void oneWayMessageTest() throws Exception {
    //创建一个生产者，并指定一个组名
    DefaultMQProducer producer = new DefaultMQProducer("oneway-producer-group");
    //连接namesrv,参数是namesrv的ip地址:端口号
    producer.setNamesrvAddr(MqConstant.NAME_SERVER_ADDR);
    //启动
    producer.start();


    //指定topic，创建一个消息
    Message message = new Message("onewayTopic1", "这是一条单向消息".getBytes());

    //发送单向消息
    producer.sendOneway(message);

    producer.shutdown();
}

发送延迟消息

消息放入mq后，过一段时间，才会被监听到，然后消费.

比如下订单业务，提交了一个订单就可以发送一个延时消息，30min后去检查这个订单的状态，如果还是未付款就取消订单释放库存。

代码如下

@Test
public void msMessageTest() throws Exception{
    //创建一个生产者，并指定一个组名
    DefaultMQProducer producer = new DefaultMQProducer("ms-producer-group");
    //连接namesrv,参数是namesrv的ip地址:端口号
    producer.setNamesrvAddr(MqConstant.NAME_SERVER_ADDR);
    //启动
    producer.start();

    //指定topic，创建一个消息
    Message message = new Message("msTopic1", "这是一条单向消息".getBytes());
    //给消息设置一个延迟时间
    message.setDelayTimeLevel(3);

    //发送延时消息
    producer.sendOneway(message);

    producer.shutdown();
}

延时等级如下：

发送批量消息

代码如下：

@Test
public void testBatchProducer() throws Exception {
    // 创建默认的生产者
    DefaultMQProducer producer = new DefaultMQProducer("test-batch-group");
    //连接namesrv,参数是namesrv的ip地址:端口号
    producer.setNamesrvAddr(MqConstant.NAME_SERVER_ADDR);
    // 启动实例
    producer.start();
    List<Message> msgs = Arrays.asList(
        new Message("batchTopicTest", "我是一组消息的A消息".getBytes()),
        new Message("batchTopicTest", "我是一组消息的B消息".getBytes()),
        new Message("batchTopicTest", "我是一组消息的C消息".getBytes())

    );
    SendResult send = producer.send(msgs);
    System.out.println(send);
    // 关闭实例
    producer.shutdown();
}

这些消息会被放到同一个队列中。

发送顺序消息

可以想象一个场景，我们在网上购物时，需要先完成下订单操作，然后再去发短信，再进行发货，需要保证顺序的。

前文我们讲的都是
并发消息
，这种消息并不能完成上述的场景逻辑。比如一个topic里有10个消息，分别在4个队列中；

• 如果消费者，同时有20个线程在消费，可能A线程拿到消息1了，B线程拿到消息2了，但是B线程可能完成的比A线程早，这就没办法上述场景的顺序了。
• 如果消费者只有一个线程，轮询消费四个队列中的消息时，也不能保证是网购场景中的顺序的。

这就要引出
顺序消息
：把消费者变为单线程，把下订单消息、发短信消息、发货消息放到同一个队列就可以了。

代码

消息封装成实体类如下：

@Data
@AllArgsConstructor
@NoArgsConstructor
public class MessageModel {
    //订单id
    private String orderId;
    //用户id
    private String userId;
    //消息描述
    private String description;
}

发送顺序消息的生产者代码如下：

/**
     * 顺序消息
     */

@Test
public void testOrderlyProducer() throws Exception {

    List<MessageModel> messageModelList = Arrays.asList(
        //用户1的订单
        new MessageModel("order-666666","user-1","下单"),
        new MessageModel("order-666666","user-1","发短信"),
        new MessageModel("order-666666","user-1","发货"),

        //用户2的订单
        new MessageModel("order-222","user-2","下单"),
        new MessageModel("order-222","user-2","发短信"),
        new MessageModel("order-222","user-2","发货")
    );

    // 创建默认的生产者
    DefaultMQProducer producer = new DefaultMQProducer("test-orderly-group");
    //连接namesrv,参数是namesrv的ip地址:端口号
    producer.setNamesrvAddr(MqConstant.NAME_SERVER_ADDR);
    // 启动实例
    producer.start();

    //发送顺序消息时 发送时相同用户的消息要保证有序，并且发到同一个队列里
    messageModelList.forEach(
        messageModel->{
            Message message = new Message("orderlyTopic", messageModel.toString().getBytes());
            try {
                //发送消息，相同订单号去相同队列
                producer.send(message, new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message message, Object arg) {
                        //producer.send(message,selector,arg),第三个参数订单号会传给selector要实现的方法的arg
                        //在这里选择队列
                        int hashCode = Math.abs(arg.toString().hashCode());
                        int index = hashCode % mqs.size();
                        return mqs.get(index);
                    }
                }, messageModel.getOrderId());
            } catch (Exception e) {
                log.error("有错误发生",e);
            }
        }
    );
    // 关闭实例
    producer.shutdown();
    log.info("发送完成");
}

消费顺序消息的消费者代码如下：

//消费者
@Test
public void orderlyConsumer() throws Exception {
    //创建一个消费者，并指定一个组名
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-orderly-consumer-group");
    //连接namesrv,参数是namesrv的ip地址:端口号
    consumer.setNamesrvAddr(MqConstant.NAME_SERVER_ADDR);
    //订阅一个主题 *号表示订阅这个主题中所有的消息
    consumer.subscribe("orderlyTopic","*");
    //设置一个监听器（一直监听，异步回调方式）
    consumer.registerMessageListener(new MessageListenerOrderly() {
        @Override
        public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
            log.info("线程id"+Thread.currentThread().getId());
            log.info("消息内容:"+new String(msgs.get(0).getBody()));
            return ConsumeOrderlyStatus.SUCCESS;
        }
    });

    //启动消费者
    consumer.start();

    //挂起当前jvm,防止主线程结束，让监听器一直监听
    System.in.read();

}

运行结果如下：

可以看到同一个订单是顺序消费的。

其他问题

如果我们的消息消费失败了怎么办？

如果是并发模式，消费失败会进行重试，重试16次后还会没消费成功，会被放到死信队列里。

如果是顺序模式，如果重试失败，会无限重试，是int的最大值。

发送带标签的消息，消息过滤

如果我们有衣服订单的消息、手机订单的消息，如果我们只使用topic进行区分，就要使用两个topic；但是它们都是订单，所以在同一个topic中会好一些，Rocketmq就提供了消息过滤功能，通过tag或者key进行区分。

生产者代码如下：

@Test
public void testTagProducer() throws Exception {
    // 创建默认的生产者
    DefaultMQProducer producer = new DefaultMQProducer("test-tag-group");
    //连接namesrv,参数是namesrv的ip地址:端口号
    producer.setNamesrvAddr(MqConstant.NAME_SERVER_ADDR);
    // 启动实例
    producer.start();
    Message messageTopic1 = new Message("tagTopic", "tag1", "这是topic1的消息".getBytes());
    Message messageTopic2 = new Message("tagTopic", "tag2", "这是topic2的消息".getBytes());
    producer.send(messageTopic1);
    producer.send(messageTopic2);
    // 关闭实例
    producer.shutdown();
}

消费tag1的消费者

//消费tag1的消费者
@Test
public void tagConsumer1() throws Exception {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tag-consumer-group-a");
    consumer.setNamesrvAddr(MqConstant.NAME_SERVER_ADDR);
    consumer.subscribe("tagTopic", "tag1");
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            System.out.println("我是tag1的消费者，我正在消费消息" + new String(msgs.get(0).getBody()));
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    consumer.start();
    System.in.read();
}

消费tag1和tag2的消费者

//消费tag1和tag2的消费者
@Test
public void tagConsumer2() throws Exception {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tag-consumer-group-a");
    consumer.setNamesrvAddr(MqConstant.NAME_SERVER_ADDR);
    consumer.subscribe("tagTopic", "tag1 || tag2");
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            System.out.println("我是tag1和tag2的消费者，我正在消费消息" + new String(msgs.get(0).getBody()));
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    consumer.start();
    System.in.read();
}

带key的消息

消息都会有自己的MessageId的，如下图：

那我们能否指定id呢？

在发送消息时可以指定key：

@Test
public void testKeyProducer() throws Exception {
    // 创建默认的生产者
    DefaultMQProducer producer = new DefaultMQProducer("test-key-group");
    //连接namesrv,参数是namesrv的ip地址:端口号
    producer.setNamesrvAddr(MqConstant.NAME_SERVER_ADDR);

    String key = UUID.randomUUID().toString();

    // 启动实例
    producer.start();
    Message messageTopic1 = new Message("keyTopic", "tag1",key, "这是topic1的消息".getBytes());
    producer.send(messageTopic1);
    // 关闭实例
    producer.shutdown();
}

消费者获取key:

@Test
public void testKeyConsumer() throws Exception {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("key-consumer-group-a");
    consumer.setNamesrvAddr(MqConstant.NAME_SERVER_ADDR);
    consumer.subscribe("keyTopic","*");
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            System.out.println("我们设置的key：" + msgs.get(0).getKeys());
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    consumer.start();
    System.in.read();
}

输出如下：

1.死锁

要提交快照的时候由于没有人取走applyCh通道里面的东西，导致死锁。

具体解释：

2D的测试代码中在日志达到一定大小时会调用snapshot，该函数需要申请rf.mu这个互斥锁。而在提交普通的日志条目时，错误地没有先释放锁，导致snapshot无法进行下去，相关的进程卡在rf.mu这个锁上，无法完成快照，更无法处理applyCh通道的下一个日志条目。这导致了向通道中提交日志条目也会因为applyCh已满而被阻塞。

2.快速定位错误

可以看到打印的日志中出现了“whe： u4"的信息，就可以推知：相关的错误发现在被u4标记的代码处，

在访问日志具体条目的代码处，传入相应的标记，这样当调用getEntry函数失败时，能快速定位目标。

3.由于截断日志增加的处理

3.1

发生u4报错，定位到相应代码。

在leader方，由于prevLogIndex处的日志条目被截取，小于rf.log.start(), 在运行getEntry函数时发生报错。

解决方法：

1.设置prevLogIndex = rf.log.start()

2.应发送给follower的日志条目被删除，直接发送快照给follower。

3.2

出错的点在于follower这边：leader方发送出去的时候prevLogIndex没有低于其自身的start，故没有发送快照，但是接收方收到日志条目之后，由于已经截断了日志，并不能匹配prevLogIndex。

显然接收方对这种情况也需要处理，并不能仅仅返回个error就完事了。

解决方案：

设置XLen为start（）+1， 即日志中的第一个条目，leader在收到回应的时候会执行nextIndex[i] = XLen, 这样就将nextIndex设置为follower方的日志第一个条目。

3.3

当prevLogIndex等于start时候，由于不匹配可能导致添加条目无法成功。

在截断日志的时候需要设置占位的条目的term为snapshot.term；无论是安装快照的时候，还是自己截断日志、生成快照的时候.

4关于应用（apply）时索引的思考

这是崩溃之前发送的消息，可以看出发送的最后一个索引号是223

这是重启之后，接收快照后而开始apply的快照，索引为229。

显然：

当start/restart后, 如果先发送的是日志条目，索引只能从1开始；但是如果是快照的话，索引可以从任意值开始，而其后的日志条目的索引值只需从该值递增即可。

一个有趣的发现：

TestSnapshotInstallUnCrash2D 每次只会使一个server崩溃，而TestSnapshotAllCrash2D 将会使得所有server同时崩溃；前者会使得commitIndex能够维持在正确值，而不回退，而后者会使得commitIndex从0开始。

5.发现不能通过从快照中恢复的测试

增加打印语句之后，发现restart之时都是没有带上快照的，这已经很说明问题了。

我并没有实现从稳定存储中读取快照的方法。

增添语句后，即可通过所有测试。

最终效果

整个实验2大致需要花费364秒左右，还是很满意的。

设计模式浅析(二) ·观察者模式

日常叨逼叨

java设计模式浅析，如果觉得对你有帮助，记得一键三连，谢谢各位观众老爷