カスタムタスクをaiogramexecutorに統合する方法は？

import asyncio
from threading import Thread
from datetime import datetime
from aiogram import Bot, Dispatcher, executor, types

API_TOKEN = ''

bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot)

chat_ids = {}

@dp.message_handler()
async def echo(message: types.Message):
    # old style:
    # await bot.send_message(message.chat.id, message.text)

    chat_ids[message.message_id] = message.from_user
    text = f'{message.message_id} {message.from_user} {message.text}'
    await message.reply(text, reply=False)


async def periodic(sleep_for, queue):
    while True:
        await asyncio.sleep(sleep_for)
        now = datetime.utcnow()
        print(f"{now}")
        for id in chat_ids:
            queue.put_nowait((id, f"{now}"))
            # await bot.send_message(id, f"{now}", disable_notification=True)


def run_tick(queue):
    newloop = asyncio.new_event_loop()
    asyncio.set_event_loop(newloop)
    asyncio.run(periodic(3, queue))


if __name__ == '__main__':
    queue = asyncio.Queue()
    Thread(target=run_tick, args=(queue,), daemon=True).start()
    executor.start_polling(dp, skip_updates=True)

イベントが発生したが今のところ失敗した場合、bot.send_messageで登録ユーザーにメッセージを送信したい。これが私が試したものです。

1. bot.send_messageは、別のスレッドから呼び出されたためにクラッシュします。（タイムアウトコンテキストマネージャーはタスク内で使用する必要があります）
2. そのため、キューを使用してこれを回避しようとしましたが、自分のタスクをエグゼキュータに追加する方法はありません。

これを行う簡単な方法はありますか？

編集：2020-1-3

@ user4815162342による作業例を次に示します。

import asyncio
from datetime import datetime
from aiogram import Bot, Dispatcher, executor, types

API_TOKEN = ''

bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot)

chat_ids = {}

@dp.message_handler()
async def echo(message: types.Message):
    chat_ids[message.from_user.id] = message.from_user
    text = f'{message.message_id} {message.from_user} {message.text}'
    await message.reply(text, reply=False)

async def periodic(sleep_for):
    while True:
        await asyncio.sleep(sleep_for)
        now = datetime.utcnow()
        print(f"{now}")
        for id in chat_ids:
            await bot.send_message(id, f"{now}", disable_notification=True)

if __name__ == '__main__':
    dp.loop.create_task(periodic(10))
    executor.start_polling(dp)