package com.lnwazg;

import java.util.List;
import java.util.Map;

import com.lnwazg.api.DistributedTask;
import com.lnwazg.bean.HandleResult;

/**
 * Первый узел считает от 1 до 10000, второй узел считает от 10001 до 20000 и так далее.
 * В итоге отправляется общая сумма данных.
 * @author nan.li
 * @version 2017年7月14日
 */
public class Task002 extends DistributedTask {
    @Override
    public void executeCustom(Map<String, Object> map) {
        // Сложение результатов
        int sum = 0;
        int start = nodeNum * 10000 + 1;//1        10001
        int end = (nodeNum + 1) * 10000;//10000     20000
        for (int i = start; i <= end; i++) {
            sum += i;
        }
        // Отправка результата вычислений
        report("sum", sum);
    }

    public String getTaskDescription() {
        return "Первый узел считает от 1 до 10000, второй узел считает от 10001 до 20000 и так далее. В итоге отправляется общая сумма данных";
    }

    @Override
    public HandleResult reducer(Map<String, List<HandleResult>> handleResultsMap) {
        int sum = 0;
        for (String nodeNum : handleResultsMap.keySet()) {
            List<HandleResult> handleResults = handleResultsMap.get(nodeNum);
            // Поскольку есть только один шаг отправки, достаточно взять результат первого шага!
            sum += Integer.valueOf(handleResults.get(0).getParamMap().get("sum"));
        }
        return new HandleResult().setResult(sum);
    }
}

package com.lnwazg.api;

import java.util.List;
import java.util.Map;

import com.lnwazg.bean.HandleResult;
import com.lnwazg.swing.util.WinMgr;
import com.lnwazg.ui.MainFrame;

/**
 * Абстрактный класс для распределенных задач, который упрощает реализацию конкретных задач.
 * @author nan.li
 * @version 2017年7月6日
 */
public abstract class DistributedTask {
    /**
     * Текущий номер узла
     */
    protected int nodeNum = 0;

    /**
     * Номер запроса текущего узла
     */
    protected int nodeNumReqNum = 0;

    /**
     * Единый вызов метода выполнения DataNode, ядро метода выполнения.
     * @param map
     */
    public void execute(Map<String, Object> map) {
        // Получить текущий номер узла
        nodeNum = Integer.valueOf(map.get("nodeNum").toString());

        // Выполнить пользовательский код
        executeCustom(map);

        // Завершить вычисление
        end();
    }

    /**
     * Пользовательский метод, который должен быть реализован клиентом.
     * @param map
     */
    public abstract void executeCustom(Map<String, Object> map);

    /**
     * mapper, обработка на стороне сервера, сохранение параметров запроса.
     * @param paramMap  Параметры таблицы, параметры, отправленные через mq.
     * @return  Объект результата обработки
     */
    public HandleResult mapper(Map<String, String> paramMap) {
        // Номер узла
        String nodeNum = paramMap.get("nodeNum");

        // Номер запроса узла
        String nodeNumReqNum = paramMap.get("nodeNumReqNum");

        // Результат одношаговой обработки
        return new HandleResult()
                .setNodeNum(nodeNum)
                .setNodeNumReqNum(nodeNumReqNum)
                .setParamMap(paramMap);
    }

    /**
     * reducer, обработка на стороне сервера.
     * @param handleResultsMap  Таблица результатов, результаты каждого шага обработки.
     * @return  Окончательный результат обработки
     */
    public HandleResult reducer(Map<String, List<HandleResult>> handleResultsMap) {
        return null;
    }

    /**
     * Получить описание задачи.
     * @return
     */
    public abstract String getTaskDescription();

    /**
     * Получите текущее имя клиента.
     * @return
     */
    public String getCurrentDataNodeName() {
        return WinMgr.win(MainFrame.class).myselfAddress;
    }

    /**
     * Отчет о данных процесса вычисления <br>
     * Будет автоматически отправлен текущий номер узла.
     * @param key
     * @param
``` **Четыре: анализ**

NameNode и DataNode — это прямые отсылки к концепциям Hadoop, но они более упрощены:
* NameNode используется для распространения исполняемых jar-пакетов, сбора промежуточных результатов вычислений и объединения итоговых результатов.
* DataNode используется для приёма исполняемых jar-пакетов, выполнения указанных методов задач и предоставления отчётов о вычислениях (промежуточных).
* MyZooKeeper используется для регистрации и поиска MQ-сервиса.

NameNode и DataNode взаимодействуют асинхронно через MQ.

Принцип работы:
1. Когда DataNode подключается к сети, он отправляет сообщение о подключении в MQ. NameNode прослушивает это сообщение и сохраняет список подключённых DataNodes.
2. Когда NameNode отправляет исполняемый jar-пакет, он последовательно отправляет сообщения каждой записи в локальном списке List<DataNode>. Сообщение содержит URL-адрес jar-пакета.
3. После получения сообщения каждый DataNode выполняет указанный метод из jar-пакета и предоставляет отчёт о промежуточных данных.
4. После завершения выполнения каждый DataNode должен выполнить метод end(), чтобы указать, что выполнение завершено.
5. Когда NameNode получает сообщения о завершении выполнения от всех DataNodes, он начинает объединять все промежуточные результаты и вычисляет итоговый результат.

**Пять: метод запуска релиза**
1. Настроить хост на локальном компьютере: 127.0.0.1 myzoo.lnwazg.com.
2. Запустить сервис MY_ZOO_SERVER-1.1.jar для регистрации и обнаружения сервера.
3. Запустить сервер MQ_SERVER-1.1.jar.
4. Запустить узел управления NameNode-1.1.jar.
5. Запустить рабочие узлы DataNode-1.1.jar, можно запустить несколько.
6. В интерфейсе NameNode-1.1.jar нажать «Загрузить задачу jar-файл и выполнить» для отправки распределённых вычислений jar-файла.

Примечание: примеры jar-файлов находятся в каталоге release/jartasks/.

**Шесть: обновление версии**
v1.1
Первая версия предоставляет только базовые функции: распространение jar-пакетов и удалённое выполнение, сервер может объединять результаты выполнения или выполнять мониторинг и анализ.

Эта версия обеспечивает основные характеристики, необходимые для распределённой вычислительной системы. На её основе можно реализовать:
* распределённый сбор данных;
* статистические расчёты;
* бизнес-операции;
* декомпозицию вычислительных задач.

Конкретные сценарии применения включают:
* сбор изображений и текста;
* расчёт значения числа π;
* управление машинами в кластере для проведения распределённых акций по продаже товаров;
* анализ сигналов из космоса и структуры биологических молекул.

В целом, конкретные функции, которые необходимо реализовать, зависят от ваших потребностей и воображения.

v1.2
Это обновление включает в себя улучшение важных функций и исправление ошибок.

Добавлены две важные функции:
* проверка состояния клиента;
* однократное обновление всех онлайн-клиентов.

Проверка состояния клиента позволяет отслеживать состояние каждого клиента в реальном времени: нормальное или отключённое.

Однократное обновление клиентов позволяет автоматически обновлять все онлайн-клиенты одним нажатием кнопки на сервере. Каждый клиент автоматически перезапускается после обновления.